5 Struct N_KAFKA
gullradriel edited this page 2026-10-09 00:55:18 +02:00

N_KAFKA

Module: KAFKA: generic event producer and consumer functions

structure of a KAFKA consumer or producer handle

#include "n_kafka.h"

Public Attributes

Name
char * tosend_dir
base directory the produced event files are read from ('to-send').
char ** topics
list of topics to subscribe to
char * topic
kafka topic string
rd_kafka_topic_partition_list_t * subscription
subscribed topics
char * sended_dir
base directory produced event files are moved to once acknowledged ('sended').
int schema_id
kafka schema id in network order
pthread_rwlock_t rwlock
access lock
LIST * received_events
list of received N_KAFKA_EVENT
rd_kafka_topic_t * rd_kafka_topic
kafka topic handle
rd_kafka_t * rd_kafka_handle
kafka handle (producer or consumer)
rd_kafka_conf_t * rd_kafka_conf
kafka structure handle
int polling_thread_status
polling thread status, 0 => off , 1 => on , 2 => wants to stop, will be turned out to 0 by exiting polling thread
pthread_t polling_thread
polling thread id
int32_t poll_timeout
poll timeout in msecs
int32_t poll_interval
poll interval in msecs
size_t nb_waiting
number of events waiting for an ack in the waiting list
size_t nb_queued
number of waiting events in the producer waiting list
size_t nb_error
number of errored events
int32_t monitored_directory_interval
monitored directory refresh interval in msecs
int mode
kafka handle mode: RD_KAFKA_CONSUMER or RD_KAFKA_PRODUCER
int is_transactional
TRUE when a 'transactional.id' is set in the producer config: produces are then wrapped in kafka transactions.
char * groupid
consumer group id
LIST * events_to_send
list of N_KAFKA_EVENT to send
int32_t event_production_enabled
bool flag to suspend or restart event production
int32_t event_consumption_enabled
bool flag to suspend or restart event consumption
char * event_cmd
eventual custom event_cmd
N_STR * errstr
kafka error string holder
int32_t error_timeout
retry interval in seconds for events stuck in N_KAFKA_EVENT_ERROR.
cJSON * configuration
kafka json configuration holder
char * bootstrap_servers
kafka bootstrap servers string

Public Attributes Documentation

variable tosend_dir

char * tosend_dir;

base directory the produced event files are read from ('to-send').

Used together with sended_dir to compute the destination subpath when moving a file on ack. NULL keeps the legacy behaviour.

variable topics

char ** topics;

list of topics to subscribe to

variable topic

char * topic;

kafka topic string

variable subscription

rd_kafka_topic_partition_list_t * subscription;

subscribed topics

variable sended_dir

char * sended_dir;

base directory produced event files are moved to once acknowledged ('sended').

When NULL, acknowledged files are deleted (legacy behaviour) instead of being moved.

variable schema_id

int schema_id;

kafka schema id in network order

variable rwlock

pthread_rwlock_t rwlock;

access lock

variable received_events

LIST * received_events;

list of received N_KAFKA_EVENT

variable rd_kafka_topic

rd_kafka_topic_t * rd_kafka_topic;

kafka topic handle

variable rd_kafka_handle

rd_kafka_t * rd_kafka_handle;

kafka handle (producer or consumer)

variable rd_kafka_conf

rd_kafka_conf_t * rd_kafka_conf;

kafka structure handle

variable polling_thread_status

int polling_thread_status;

polling thread status, 0 => off , 1 => on , 2 => wants to stop, will be turned out to 0 by exiting polling thread

variable polling_thread

pthread_t polling_thread;

polling thread id

variable poll_timeout

int32_t poll_timeout;

poll timeout in msecs

variable poll_interval

int32_t poll_interval;

poll interval in msecs

variable nb_waiting

size_t nb_waiting;

number of events waiting for an ack in the waiting list

variable nb_queued

size_t nb_queued;

number of waiting events in the producer waiting list

variable nb_error

size_t nb_error;

number of errored events

variable monitored_directory_interval

int32_t monitored_directory_interval;

monitored directory refresh interval in msecs

variable mode

int mode;

kafka handle mode: RD_KAFKA_CONSUMER or RD_KAFKA_PRODUCER

variable is_transactional

int is_transactional;

TRUE when a 'transactional.id' is set in the producer config: produces are then wrapped in kafka transactions.

FALSE keeps the legacy fire-and-forget produce path.

variable groupid

char * groupid;

consumer group id

variable events_to_send

LIST * events_to_send;

list of N_KAFKA_EVENT to send

variable event_production_enabled

int32_t event_production_enabled;

bool flag to suspend or restart event production

variable event_consumption_enabled

int32_t event_consumption_enabled;

bool flag to suspend or restart event consumption

variable event_cmd

char * event_cmd;

eventual custom event_cmd

variable errstr

N_STR * errstr;

kafka error string holder

variable error_timeout

int32_t error_timeout;

retry interval in seconds for events stuck in N_KAFKA_EVENT_ERROR.

0 (default) disables retries, errored events stay errored until the caller dumps and reloads them.

variable configuration

cJSON * configuration;

kafka json configuration holder

variable bootstrap_servers

char * bootstrap_servers;

kafka bootstrap servers string