src/n_kafka.c

Generic kafka consume and produce event functions. More...

Functions

	Name
int	n_kafka_produce_ex(N_KAFKA * kafka, N_KAFKA_EVENT * event) produce an event on a N_KAFKA *kafka handle
void *	n_kafka_polling_thread(void * ptr) kafka produce or consume polling thread function
int	n_kafka_move_file_to_sended(const N_KAFKA * kafka, const char * filepath) move a single acknowledged event file to the 'sended' directory, preserving its subpath relative to 'to-send'
int	n_kafka_mkdir_p(const char * path) create a directory and all its missing parents (mkdir -p)
const char *	n_kafka_event_fdesc(const N_KAFKA_EVENT * event) return a " (file: ...)" log fragment naming the source file(s) of an event
void	n_kafka_delivery_message_callback(rd_kafka_t * rk, const rd_kafka_message_t * rkmessage, void * opaque) Message delivery report callback.

Detailed Description

Generic kafka consume and produce event functions.

Author: Castagnier Mickael

Version: 1.0

Date: 23/11/2022

Functions Documentation

function n_kafka_produce_ex

int n_kafka_produce_ex(
    N_KAFKA * kafka,
    N_KAFKA_EVENT * event
)

produce an event on a N_KAFKA *kafka handle

Parameters:

• kafka the producer handle to use
• event event to send

Return: TRUE or FALSE

function n_kafka_polling_thread

void * n_kafka_polling_thread(
    void * ptr
)

kafka produce or consume polling thread function

Parameters:

• ptr (void *)kafka handle

Return: NULL

function n_kafka_move_file_to_sended

static int n_kafka_move_file_to_sended(
    const N_KAFKA * kafka,
    const char * filepath
)

move a single acknowledged event file to the 'sended' directory, preserving its subpath relative to 'to-send'

Parameters:

• kafka the producer handle holding tosend_dir/sended_dir
• filepath the full path of the file to move (as found under tosend_dir)

Return: TRUE on success, FALSE on error

function n_kafka_mkdir_p

static int n_kafka_mkdir_p(
    const char * path
)

create a directory and all its missing parents (mkdir -p)

Parameters:

• path the directory path to create

Return: TRUE on success (or if it already exists), FALSE on error

function n_kafka_event_fdesc

static const char * n_kafka_event_fdesc(
    const N_KAFKA_EVENT * event
)

return a " (file: ...)" log fragment naming the source file(s) of an event

Parameters:

• event the event to describe

Return: a never-NULL const C string to append after the opaque pointer in a log line. When the event carries its source file(s) (event_files_to_delete, set when the event is produced from a monitored directory) the fragment names them; for events with no source file (string events, single-file sends, received events) it is empty so no misleading "(file: ...)" is printed.

function n_kafka_delivery_message_callback

static void n_kafka_delivery_message_callback(
    rd_kafka_t * rk,
    const rd_kafka_message_t * rkmessage,
    void * opaque
)

Message delivery report callback.

Parameters:

• rk kafka rd_kafka_t kafka handle
• rkmessage pointer to the received event in kafka form
• opaque opaque structure holding the pointer of a sent event

This callback is called exactly once per message, indicating if the message was succesfully delivered The callback is triggered from rd_kafka_poll() and executes on the application's thread

Source code

/*
 * Nilorea Library
 * Copyright (C) 2005-2026 Castagnier Mickael
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
 * implied. See the License for the specific language governing
 * permissions and limitations under the License.
 *
 * SPDX-License-Identifier: Apache-2.0
 */


#include "nilorea/n_kafka.h"
#include "nilorea/n_common.h"
#include "nilorea/n_base64.h"
#include <limits.h>
#include <sys/stat.h>
#include <sys/types.h>

static const char* n_kafka_event_fdesc(const N_KAFKA_EVENT* event) {
    static _Thread_local char fragment[2 * PATH_MAX];
    if (event && event->event_files_to_delete && event->event_files_to_delete->data) {
        snprintf(fragment, sizeof(fragment), " (file: %s)", event->event_files_to_delete->data);
        return fragment;
    }
    return "";
}

static int n_kafka_mkdir_p(const char* path) {
    __n_assert(path, return FALSE);

    size_t len = strlen(path);
    if (len == 0)
        return FALSE;

    char* tmp = strdup(path);
    __n_assert(tmp, return FALSE);

    /* strip a trailing slash so the final component is created too */
    if (tmp[len - 1] == '/')
        tmp[len - 1] = '\0';

    for (char* p = tmp + 1; *p; p++) {
        if (*p == '/') {
            *p = '\0';
            if (mkdir(tmp, 0775) != 0 && errno != EEXIST) {
                int error = errno;
                n_log(LOG_ERR, "could not create directory \"%s\": %s", tmp, strerror(error));
                free(tmp);
                return FALSE;
            }
            *p = '/';
        }
    }
    if (mkdir(tmp, 0775) != 0 && errno != EEXIST) {
        int error = errno;
        n_log(LOG_ERR, "could not create directory \"%s\": %s", tmp, strerror(error));
        free(tmp);
        return FALSE;
    }
    free(tmp);
    return TRUE;
}

static int n_kafka_move_file_to_sended(const N_KAFKA* kafka, const char* filepath) {
    __n_assert(kafka, return FALSE);
    __n_assert(kafka->sended_dir, return FALSE);
    __n_assert(filepath, return FALSE);

    /* compute the path relative to the 'to-send' base directory; fall back to
     * the basename when the file does not live under tosend_dir */
    const char* relative = NULL;
    if (kafka->tosend_dir) {
        size_t base_len = strlen(kafka->tosend_dir);
        if (strncmp(filepath, kafka->tosend_dir, base_len) == 0) {
            relative = filepath + base_len;
            while (*relative == '/')
                relative++;
        }
    }
    N_STR* dest = NULL;
    if (relative && *relative) {
        nstrprintf(dest, "%s/%s", kafka->sended_dir, relative);
    } else {
        char* tmp = strdup(filepath);
        __n_assert(tmp, return FALSE);
        nstrprintf(dest, "%s/%s", kafka->sended_dir, basename(tmp));
        free(tmp);
    }
    __n_assert(dest, return FALSE);

    /* ensure the destination directory exists */
    char* dest_copy = strdup(_nstr(dest));
    if (dest_copy) {
        n_kafka_mkdir_p(dirname(dest_copy));
        free(dest_copy);
    }

    int ret = TRUE;
    if (rename(filepath, _nstr(dest)) != 0) {
        int error = errno;
        n_log(LOG_ERR, "could not move acknowledged file \"%s\" to \"%s\": %s", filepath, _nstr(dest), strerror(error));
        ret = FALSE;
    } else {
        n_log(LOG_DEBUG, "moved acknowledged file \"%s\" to \"%s\"", filepath, _nstr(dest));
    }
    free_nstr(&dest);
    return ret;
}

int32_t n_kafka_get_schema_from_char(const char* string) {
    __n_assert(string, return -1);

    uint32_t raw_schema_id = 0;
    memcpy(&raw_schema_id, string + 1, sizeof(uint32_t));

    return (int32_t)ntohl(raw_schema_id);
}

int32_t n_kafka_get_schema_from_nstr(const N_STR* string) {
    __n_assert(string, return -1);
    __n_assert(string->data, return -1);
    __n_assert((string->written >= sizeof(int32_t)), return -1);
    return n_kafka_get_schema_from_char(string->data);
}

int n_kafka_put_schema_in_char(char* string, int schema_id) {
    __n_assert(string, return FALSE);
    uint32_t schema_id_htonl = htonl((uint32_t)schema_id);  // cast to unsigned to avoid warning
    memcpy(string + 1, &schema_id_htonl, sizeof(uint32_t));
    return TRUE;
}

int n_kafka_put_schema_in_nstr(N_STR* string, int schema_id) {
    __n_assert(string, return FALSE);
    __n_assert(string->data, return FALSE);
    __n_assert((string->written >= sizeof(int32_t)), return FALSE);
    return n_kafka_put_schema_in_char(string->data, schema_id);
}

int n_kafka_get_status(N_KAFKA* kafka, size_t* nb_queued, size_t* nb_waiting, size_t* nb_error) {
    __n_assert(kafka, return FALSE);
    __n_assert(nb_waiting, return FALSE);
    __n_assert(nb_error, return FALSE);
    read_lock(kafka->rwlock);
    int status = kafka->polling_thread_status;
    *nb_queued = kafka->nb_queued;
    *nb_waiting = kafka->nb_waiting;
    *nb_error = kafka->nb_error;
    unlock(kafka->rwlock);
    return status;
}

int n_kafka_set_error_timeout(N_KAFKA* kafka, int32_t error_timeout) {
    __n_assert(kafka, return FALSE);
    if (error_timeout < 0) {
        n_log(LOG_ERR, "n_kafka_set_error_timeout: negative timeout %d rejected", error_timeout);
        return FALSE;
    }
    write_lock(kafka->rwlock);
    kafka->error_timeout = error_timeout;
    unlock(kafka->rwlock);
    return TRUE;
}

int n_kafka_set_sended_dir(N_KAFKA* kafka, const char* tosend_dir, const char* sended_dir) {
    __n_assert(kafka, return FALSE);
    write_lock(kafka->rwlock);
    FreeNoLog(kafka->tosend_dir);
    FreeNoLog(kafka->sended_dir);
    if (tosend_dir)
        kafka->tosend_dir = strdup(tosend_dir);
    if (sended_dir)
        kafka->sended_dir = strdup(sended_dir);
    unlock(kafka->rwlock);
    return TRUE;
}

// cppcheck-suppress[constParameterCallback,constParameter] ; callback signature must match rd_kafka typedef
static void n_kafka_delivery_message_callback(rd_kafka_t* rk, const rd_kafka_message_t* rkmessage, void* opaque) {
    (void)opaque;

    __n_assert(rk, n_log(LOG_ERR, "rk=NULL is not a valid kafka handle"); return);
    __n_assert(rkmessage, n_log(LOG_ERR, "rkmessage=NULL is not a valid kafka message"); return);

    N_KAFKA_EVENT* event = (N_KAFKA_EVENT*)rkmessage->_private;

    if (rkmessage->err) {
        n_log(LOG_ERR, "message delivery failed for event %p%s: %s", (void*)event, n_kafka_event_fdesc(event), rd_kafka_err2str(rkmessage->err));
        if (!event) {
            n_log(LOG_ERR, "fatal: event is NULL");
            return;
        }
        if (event->parent_table)
            write_lock(event->parent_table->rwlock);
        event->status = N_KAFKA_EVENT_ERROR;
        event->error_time = time(NULL);
        /* adjust counters: event was WAITING_ACK, now it's ERROR */
        if (event->parent_table) {
            event->parent_table->nb_waiting--;
            event->parent_table->nb_error++;
        }
        if (event->parent_table)
            unlock(event->parent_table->rwlock);
    } else {
        n_log(LOG_DEBUG, "message delivered (%ld bytes, partition %d)", rkmessage->len, rkmessage->partition);
        if (event) {
            // lock
            if (event->parent_table)
                write_lock(event->parent_table->rwlock);
            // finalize produce event linked files: move them to the 'sended'
            // directory when one is configured, otherwise delete them (legacy)
            if (event->event_files_to_delete) {
                int move_to_sended = (event->parent_table && event->parent_table->sended_dir) ? 1 : 0;
                char** files_to_delete = split(_nstr(event->event_files_to_delete), ";", 0);
                if (files_to_delete) {
                    if (split_count(files_to_delete) > 0) {
                        int index = 0;
                        while (files_to_delete[index]) {
                            if (move_to_sended) {
                                n_kafka_move_file_to_sended(event->parent_table, files_to_delete[index]);
                            } else {
                                int ret = unlink(files_to_delete[index]);
                                int error = errno;
                                if (ret == 0) {
                                    n_log(LOG_DEBUG, "deleted on produce ack: %s", files_to_delete[index]);
                                } else {
                                    n_log(LOG_ERR, "couldn't delete \"%s\": %s", files_to_delete[index], strerror(error));
                                }
                            }
                            index++;
                        }
                    } else {
                        n_log(LOG_ERR, "split result is empty !");
                    }
                }
                free_split_result(&files_to_delete);
            }
            // set status
            event->status = N_KAFKA_EVENT_OK;
            // unlock
            if (event->parent_table)
                unlock(event->parent_table->rwlock);
            n_log(LOG_INFO, "kafka event %p%s received an ack !", (void*)event, n_kafka_event_fdesc(event));
        } else {
            n_log(LOG_ERR, "fatal: event is NULL");
        }
    }
    return;
    /* The rkmessage is destroyed automatically by librdkafka */
}

void n_kafka_delete(N_KAFKA* kafka) {
    __n_assert(kafka, return);

    FreeNoLog(kafka->topic);
    free_split_result(&kafka->topics);
    FreeNoLog(kafka->event_cmd);
    FreeNoLog(kafka->groupid);
    FreeNoLog(kafka->tosend_dir);
    FreeNoLog(kafka->sended_dir);

    n_kafka_stop_polling_thread(kafka);

    /* rd_kafka_topic_destroy must be called before rd_kafka_destroy */
    if (kafka->rd_kafka_topic)
        rd_kafka_topic_destroy(kafka->rd_kafka_topic);
    kafka->rd_kafka_topic = NULL;

    if (kafka->rd_kafka_handle) {
        if (kafka->mode == RD_KAFKA_CONSUMER) {
            /* close the consumer. May already be closed by stop_polling_thread,
             * calling it again is safe (returns an error but does not crash) */
            rd_kafka_consumer_close(kafka->rd_kafka_handle);
            if (kafka->subscription)
                rd_kafka_topic_partition_list_destroy(kafka->subscription);
        }
        if (kafka->mode == RD_KAFKA_PRODUCER) {
            rd_kafka_flush(kafka->rd_kafka_handle, kafka->poll_timeout);
        }
        rd_kafka_destroy(kafka->rd_kafka_handle);
        n_log(LOG_DEBUG, "kafka handle destroyed");
    }

    if (kafka->rd_kafka_conf) {
        rd_kafka_conf_destroy(kafka->rd_kafka_conf);
    }

    if (kafka->configuration)
        cJSON_Delete(kafka->configuration);

    if (kafka->errstr)
        free_nstr(&kafka->errstr);

    if (kafka->events_to_send)
        list_destroy(&kafka->events_to_send);

    if (kafka->received_events)
        list_destroy(&kafka->received_events);

    rw_lock_destroy(kafka->rwlock);

    Free(kafka->bootstrap_servers);

    Free(kafka);
    return;
}

N_KAFKA* n_kafka_new(int32_t poll_timeout, int32_t poll_interval, size_t errstr_len) {
    N_KAFKA* kafka = NULL;
    Malloc(kafka, N_KAFKA, 1);
    __n_assert(kafka, return NULL);

    kafka->errstr = new_nstr(errstr_len);
    __n_assert(kafka->errstr, Free(kafka); return NULL);

    kafka->events_to_send = NULL;
    kafka->received_events = NULL;
    kafka->rd_kafka_conf = NULL;
    kafka->rd_kafka_handle = NULL;
    kafka->configuration = NULL;
    kafka->groupid = NULL;
    kafka->topics = NULL;
    kafka->event_cmd = NULL;
    kafka->subscription = NULL;
    kafka->topic = NULL;
    kafka->mode = -1;
    kafka->schema_id = -1;
    kafka->poll_timeout = poll_timeout;
    kafka->poll_interval = poll_interval;
    kafka->monitored_directory_interval = 3000;  // 3000 msecs as a default
    kafka->nb_queued = 0;
    kafka->nb_waiting = 0;
    kafka->nb_error = 0;
    kafka->error_timeout = 0;
    kafka->event_consumption_enabled = TRUE;
    kafka->event_production_enabled = TRUE;
    kafka->polling_thread_status = 0;
    kafka->bootstrap_servers = NULL;
    kafka->is_transactional = FALSE;
    kafka->tosend_dir = NULL;
    kafka->sended_dir = NULL;

    kafka->events_to_send = new_generic_list(MAX_LIST_ITEMS);
    __n_assert(kafka->events_to_send, n_kafka_delete(kafka); return NULL);

    kafka->received_events = new_generic_list(MAX_LIST_ITEMS);
    __n_assert(kafka->received_events, n_kafka_delete(kafka); return NULL);

    if (init_lock(kafka->rwlock) != 0) {
        n_log(LOG_ERR, "could not init kafka rwlock in kafka structure at address %p", kafka);
        n_kafka_delete(kafka);
        return NULL;
    }

    return kafka;
}

N_KAFKA* n_kafka_load_config(char* config_file, int mode) {
    __n_assert(config_file, return NULL);

    N_KAFKA* kafka = NULL;

    kafka = n_kafka_new(-1, 100, 1024);
    __n_assert(kafka, return NULL);

    // initialize kafka object
    kafka->rd_kafka_conf = rd_kafka_conf_new();

    // load config file
    N_STR* config_string = NULL;
    config_string = file_to_nstr(config_file);
    if (!config_string) {
        n_log(LOG_ERR, "unable to read config from file %s !", config_file);
        n_kafka_delete(kafka);
        return NULL;
    }
    cJSON* json = NULL;
    json = cJSON_Parse(_nstrp(config_string));
    /* config_string is no longer needed after parsing */
    free_nstr(&config_string);
    if (!json) {
        n_log(LOG_ERR, "unable to parse json from file %s", config_file);
        n_kafka_delete(kafka);
        return NULL;
    }

    int jsonIndex;
    for (jsonIndex = 0; jsonIndex < cJSON_GetArraySize(json); jsonIndex++) {
        cJSON* entry = cJSON_GetArrayItem(json, jsonIndex);

        if (!entry) continue;
        __n_assert(entry->string, continue);
        if (!entry->valuestring) {
            n_log(LOG_DEBUG, "no valuestring for entry %s", _str(entry->string));
            continue;
        }

        if (entry->string[0] != '-') {
            // if it's not one of the optionnal parameters not managed by kafka, then we can use rd_kafka_conf_set on them
            if (strcmp("topic", entry->string) != 0 &&
                strcmp("topics", entry->string) != 0 &&
                strcmp("event_cmd", entry->string) != 0 &&
                strcmp("value.schema.id", entry->string) != 0 &&
                strcmp("value.schema.type", entry->string) != 0 &&
                strcmp("poll.interval", entry->string) != 0 &&
                strcmp("poll.timeout", entry->string) != 0 &&
                strcmp("group.id.autogen", entry->string) != 0 &&
                strcmp("monitored.directory.interval", entry->string) != 0) {
                if (!strcmp("group.id", entry->string)) {
                    // exclude group id for producer
                    if (mode == RD_KAFKA_PRODUCER)
                        continue;
                    kafka->groupid = strdup(entry->valuestring);
                }

                if (rd_kafka_conf_set(kafka->rd_kafka_conf, entry->string, entry->valuestring, _nstr(kafka->errstr), kafka->errstr->length) != RD_KAFKA_CONF_OK) {
                    n_log(LOG_ERR, "kafka config: %s", _nstr(kafka->errstr));
                } else {
                    n_log(LOG_DEBUG, "kafka config enabled: %s => %s", entry->string, entry->valuestring);
                }
            }
        } else {
            n_log(LOG_DEBUG, "kafka disabled config: %s => %s", entry->string, entry->valuestring);
        }
    }

    // other parameters, not directly managed by kafka API (will cause an error if used along rd_kafka_conf_set )
    // producer topic
    cJSON* jstr = NULL;
    jstr = cJSON_GetObjectItem(json, "topic");
    if (jstr && jstr->valuestring) {
        kafka->topic = strdup(jstr->valuestring);
        n_log(LOG_DEBUG, "kafka producer topic: %s", kafka->topic);
    } else {
        if (mode == RD_KAFKA_PRODUCER) {
            n_log(LOG_ERR, "no topic configured !");
            cJSON_Delete(json);
            n_kafka_delete(kafka);
            return NULL;
        }
    }
    // consumer topics
    jstr = cJSON_GetObjectItem(json, "topics");
    if (jstr && jstr->valuestring) {
        kafka->topics = split(jstr->valuestring, ",", 0);
        n_log(LOG_DEBUG, "kafka consumer topics: %s", jstr->valuestring);
    } else {
        if (mode == RD_KAFKA_CONSUMER) {
            n_log(LOG_ERR, "no topics configured !");
            cJSON_Delete(json);
            n_kafka_delete(kafka);
            return NULL;
        }
    }

    // eventual consumer custom event_cmd
    jstr = cJSON_GetObjectItem(json, "event_cmd");
    if (jstr && jstr->valuestring) {
        kafka->event_cmd = strdup(jstr->valuestring);
        n_log(LOG_DEBUG, "kafka consumer event_cmd: %s", jstr->valuestring);
    }

    // schema id if any
    jstr = cJSON_GetObjectItem(json, "value.schema.id");
    if (jstr && jstr->valuestring) {
        int schem_v = atoi(jstr->valuestring);
        if (schem_v < -1 || schem_v > 9999) {
            n_log(LOG_ERR, "invalid schema id %d", schem_v);
            cJSON_Delete(json);
            n_kafka_delete(kafka);
            return NULL;
        }
        n_log(LOG_DEBUG, "kafka schema id: %d", schem_v);
        kafka->schema_id = schem_v;
    }

    // kafka broker poll interval
    jstr = cJSON_GetObjectItem(json, "poll.interval");
    if (jstr && jstr->valuestring) {
        kafka->poll_interval = atoi(jstr->valuestring);
        n_log(LOG_DEBUG, "kafka poll interval: %d", kafka->poll_interval);
    }

    // kafka broker poll timeout
    jstr = cJSON_GetObjectItem(json, "poll.timeout");
    if (jstr && jstr->valuestring) {
        kafka->poll_timeout = atoi(jstr->valuestring);
        n_log(LOG_DEBUG, "kafka poll timeout: %d", kafka->poll_timeout);
    }

    // local directory poll interval
    jstr = cJSON_GetObjectItem(json, "monitored.directory.interval");
    if (jstr && jstr->valuestring) {
        kafka->monitored_directory_interval = atoi(jstr->valuestring);
        n_log(LOG_DEBUG, "kafka monitored directory interval: %d", kafka->monitored_directory_interval);
    }

    // bootstrap servers
    jstr = cJSON_GetObjectItem(json, "bootstrap.servers");
    if (jstr && jstr->valuestring) {
        kafka->bootstrap_servers = strdup(jstr->valuestring);
        n_log(LOG_DEBUG, "kafka bootstrap server: %s", kafka->bootstrap_servers);
    }

    if (mode == RD_KAFKA_PRODUCER) {
        // a configured 'transactional.id' switches the producer to transactional mode.
        // the key itself is already forwarded to rd_kafka_conf_set by the generic
        // loop above (it is a regular librdkafka config key), we only detect it here.
        jstr = cJSON_GetObjectItem(json, "transactional.id");
        if (jstr && jstr->valuestring && jstr->valuestring[0] != '\0') {
            kafka->is_transactional = TRUE;
            n_log(LOG_DEBUG, "kafka producer transactional.id: %s, enabling transactional produces", jstr->valuestring);
        }

        // set delivery callback
        rd_kafka_conf_set_dr_msg_cb(kafka->rd_kafka_conf, n_kafka_delivery_message_callback);

        kafka->rd_kafka_handle = rd_kafka_new(RD_KAFKA_PRODUCER, kafka->rd_kafka_conf, _nstr(kafka->errstr), kafka->errstr->length);
        if (!kafka->rd_kafka_handle) {
            n_log(LOG_ERR, "failed to create new producer: %s", _nstr(kafka->errstr));
            cJSON_Delete(json);
            n_kafka_delete(kafka);
            return NULL;
        }
        // conf is now owned by kafka handle
        kafka->rd_kafka_conf = NULL;
        // Create topic object
        kafka->rd_kafka_topic = rd_kafka_topic_new(kafka->rd_kafka_handle, kafka->topic, NULL);
        if (!kafka->rd_kafka_topic) {
            cJSON_Delete(json);
            n_kafka_delete(kafka);
            return NULL;
        }
        // initialize the transactional producer (acquires the producer id/epoch).
        // -1 lets librdkafka use 2 * transaction.timeout.ms as the blocking timeout.
        if (kafka->is_transactional) {
            rd_kafka_error_t* txn_err = rd_kafka_init_transactions(kafka->rd_kafka_handle, -1);
            if (txn_err) {
                n_log(LOG_ERR, "failed to init transactions for producer %p, topic %s: %s", kafka->rd_kafka_handle, kafka->topic, rd_kafka_error_string(txn_err));
                rd_kafka_error_destroy(txn_err);
                cJSON_Delete(json);
                n_kafka_delete(kafka);
                return NULL;
            }
            n_log(LOG_DEBUG, "transactions initialized for producer %p, topic %s", kafka->rd_kafka_handle, kafka->topic);
        }
        kafka->mode = RD_KAFKA_PRODUCER;
    } else if (mode == RD_KAFKA_CONSUMER) {
        /* If there is no previously committed offset for a partition
         * the auto.offset.reset strategy will be used to decide where
         * in the partition to start fetching messages.
         * By setting this to earliest the consumer will read all messages
         * in the partition if there was no previously committed offset. */
        /*if (rd_kafka_conf_set(kafka -> rd_kafka_conf, "auto.offset.reset", "earliest", _nstr( kafka -> errstr ) , kafka -> errstr -> length) != RD_KAFKA_CONF_OK) {
          n_log( LOG_ERR  , "kafka conf set: %s", kafka -> errstr);
          n_kafka_delete( kafka );
          return NULL ;
          }*/

        // if groupid is not set, generate a unique one
        if (!kafka->groupid) {
            char computer_name[1024] = "";
            get_computer_name(computer_name, 1024);
            N_STR* groupid = new_nstr(1024);
            char* topics = join(kafka->topics, "_");
            // generating group id
            jstr = cJSON_GetObjectItem(json, "group.id.autogen");
            if (jstr && jstr->valuestring) {
                if (strcmp(jstr->valuestring, "host-topic-group") == 0) {
                    // nstrprintf( groupid , "%s_%s_%s" , computer_name , topics ,kafka -> bootstrap_servers );
                    nstrprintf(groupid, "%s_%s", computer_name, topics);
                } else if (strcmp(jstr->valuestring, "unique-group") == 0) {
                    // nstrprintf( groupid , "%s_%s_%s_%d" , computer_name , topics , kafka -> bootstrap_servers , getpid() );
                    nstrprintf(groupid, "%s_%s_%d", computer_name, topics, getpid());
                }
            } else  // default unique group
            {
                // nstrprintf( groupid , "%s_%s_%s_%d" , computer_name , topics , kafka -> bootstrap_servers , getpid() );
                nstrprintf(groupid, "%s_%s_%d", computer_name, topics, getpid());
                n_log(LOG_DEBUG, "group.id is not set and group.id.autogen is not set, generated unique group id: %s", _nstr(groupid));
            }
            free(topics);
            kafka->groupid = groupid->data;
            groupid->data = NULL;
            free_nstr(&groupid);
        }
        if (rd_kafka_conf_set(kafka->rd_kafka_conf, "group.id", kafka->groupid, _nstr(kafka->errstr), kafka->errstr->length) != RD_KAFKA_CONF_OK) {
            n_log(LOG_ERR, "kafka consumer group.id error: %s", _nstr(kafka->errstr));
            cJSON_Delete(json);
            n_kafka_delete(kafka);
            return NULL;
        } else {
            n_log(LOG_DEBUG, "kafka consumer group.id => %s", kafka->groupid);
        }

        /* Create consumer instance.
         * NOTE: rd_kafka_new() takes ownership of the conf object
         *       and the application must not reference it again after
         *       this call.
         */
        kafka->rd_kafka_handle = rd_kafka_new(RD_KAFKA_CONSUMER, kafka->rd_kafka_conf, _nstr(kafka->errstr), kafka->errstr->length);
        if (!kafka->rd_kafka_handle) {
            n_log(LOG_ERR, "%% Failed to create new consumer: %s", _nstr(kafka->errstr));
            cJSON_Delete(json);
            n_kafka_delete(kafka);
            return NULL;
        }
        // conf is now owned by kafka handle
        kafka->rd_kafka_conf = NULL;

        /* Redirect all messages from per-partition queues to
         * the main queue so that messages can be consumed with one
         * call from all assigned partitions.
         *
         * The alternative is to poll the main queue (for events)
         * and each partition queue separately, which requires setting
         * up a rebalance callback and keeping track of the assignment:
         * but that is more complex and typically not recommended. */
        rd_kafka_poll_set_consumer(kafka->rd_kafka_handle);

        /* Convert the list of topics to a format suitable for librdkafka */
        int topic_cnt = split_count(kafka->topics);
        kafka->subscription = rd_kafka_topic_partition_list_new(topic_cnt);
        for (int i = 0; i < topic_cnt; i++)
            rd_kafka_topic_partition_list_add(kafka->subscription, kafka->topics[i],
                                              /* the partition is ignored
                                               * by subscribe() */
                                              RD_KAFKA_PARTITION_UA);
        /* Assign the topic. This method is disabled as it does noat allow dynamic partition assignement
           int err = rd_kafka_assign(kafka -> rd_kafka_handle, kafka -> subscription);
           if( err )
           {
           n_log( LOG_ERR , "kafka consumer: failed to assign %d topics: %s", kafka -> subscription->cnt, rd_kafka_err2str(err));
           n_kafka_delete( kafka );
           return NULL;
           } */

        /* Subscribe to the list of topics */
        int err = rd_kafka_subscribe(kafka->rd_kafka_handle, kafka->subscription);
        if (err) {
            n_log(LOG_ERR, "kafka consumer: failed to subscribe to %d topics: %s", kafka->subscription->cnt, rd_kafka_err2str(err));
            cJSON_Delete(json);
            n_kafka_delete(kafka);
            return NULL;
        }

        n_log(LOG_DEBUG, "kafka consumer created and subscribed to %d topic(s), waiting for rebalance and messages...", kafka->subscription->cnt);

        kafka->mode = RD_KAFKA_CONSUMER;
    } else {
        n_log(LOG_ERR, "invalid mode %d", mode);
        cJSON_Delete(json);
        n_kafka_delete(kafka);
        return NULL;
    }

    cJSON_Delete(json);
    return kafka;
} /* n_kafka_load_config */

N_KAFKA_EVENT* n_kafka_new_event(int schema_id) {
    N_KAFKA_EVENT* event = NULL;
    Malloc(event, N_KAFKA_EVENT, 1);
    __n_assert(event, return NULL);

    event->event_string = NULL;
    event->event_files_to_delete = NULL;
    event->from_topic = NULL;
    event->rd_kafka_headers = NULL;
    event->received_headers = NULL;
    event->schema_id = schema_id;
    event->status = N_KAFKA_EVENT_CREATED;
    event->parent_table = NULL;
    event->error_time = 0;

    return event;
} /* n_kafka_new_event */

int n_kafka_new_headers(N_KAFKA_EVENT* event, size_t count) {
    __n_assert(event, return FALSE);

    if (count == 0)
        count = 1;

    if (!event->rd_kafka_headers) {
        event->rd_kafka_headers = rd_kafka_headers_new(count);
        __n_assert(event->rd_kafka_headers, return FALSE);
    } else {
        n_log(LOG_ERR, "event headers already allocated for event %p%s", event, n_kafka_event_fdesc(event));
        return FALSE;
    }
    return TRUE;
}

int n_kafka_add_header_ex(N_KAFKA_EVENT* event, char* key, size_t key_length, char* value, size_t value_length) {
    __n_assert(event, return FALSE);
    __n_assert(event->rd_kafka_headers, return FALSE);
    __n_assert(key, return FALSE);
    __n_assert(value, return FALSE);

    if (key_length < 1 || key_length > SSIZE_MAX) {
        n_log(LOG_ERR, "Invalid key length (%zu) for header in event %p%s", key_length, event, n_kafka_event_fdesc(event));
        return FALSE;
    }

    if (value_length < 1 || value_length > SSIZE_MAX) {
        n_log(LOG_ERR, "Invalid value length (%zu) for key '%s' in event %p%s", value_length, key, event, n_kafka_event_fdesc(event));
        return FALSE;
    }

    rd_kafka_resp_err_t err = rd_kafka_header_add(event->rd_kafka_headers, key, (ssize_t)key_length, value, (ssize_t)value_length);

    if (err) {
        n_log(LOG_ERR, "Failed to add header [%s:%zu=%s:%zu] to event %p%s: %s",
              key, key_length, value, value_length, event, n_kafka_event_fdesc(event), rd_kafka_err2str(err));
        return FALSE;
    }

    return TRUE;
}

int n_kafka_add_header(N_KAFKA_EVENT* event, N_STR* key, N_STR* value) {
    __n_assert(event, return FALSE);
    __n_assert(key, return FALSE);
    __n_assert(key->data, return FALSE);
    __n_assert(value, return FALSE);
    __n_assert(value->data, return FALSE);

    return n_kafka_add_header_ex(event, key->data, key->written, value->data, value->written);
}

int n_kafka_produce(N_KAFKA* kafka, N_KAFKA_EVENT* event) {
    __n_assert(kafka, return FALSE);
    __n_assert(event, return FALSE);

    event->parent_table = kafka;

    write_lock(kafka->rwlock);
    kafka->nb_queued++;
    list_push(kafka->events_to_send, event, &n_kafka_event_destroy_ptr);
    unlock(kafka->rwlock);

    // Success to *enqueue* event
    n_log(LOG_DEBUG, "successfully enqueued event %p%s in producer %p waitlist, topic: %s", event, n_kafka_event_fdesc(event), kafka->rd_kafka_handle, kafka->topic);

    return TRUE;
} /* n_kafka_produce */

int n_kafka_produce_ex(N_KAFKA* kafka, N_KAFKA_EVENT* event) {
    char* event_string = NULL;
    size_t event_length = 0;

    event->parent_table = kafka;

    event_string = event->event_string->data;
    event_length = event->event_string->written;

    if (event->rd_kafka_headers) {
        rd_kafka_headers_t* hdrs_copy;
        hdrs_copy = rd_kafka_headers_copy(event->rd_kafka_headers);

        rd_kafka_resp_err_t err = rd_kafka_producev(
            kafka->rd_kafka_handle, RD_KAFKA_V_RKT(kafka->rd_kafka_topic),
            RD_KAFKA_V_PARTITION(RD_KAFKA_PARTITION_UA),
            RD_KAFKA_V_MSGFLAGS(RD_KAFKA_MSG_F_COPY),
            RD_KAFKA_V_VALUE(event_string, event_length),
            RD_KAFKA_V_HEADERS(hdrs_copy),
            RD_KAFKA_V_OPAQUE((void*)event),
            RD_KAFKA_V_END);

        if (err) {
            rd_kafka_headers_destroy(hdrs_copy);
            event->status = N_KAFKA_EVENT_ERROR;
            event->error_time = time(NULL);
            n_log(LOG_ERR, "failed to produce event: %p%s with headers %p, producer: %p, topic: %s, error: %s", event, n_kafka_event_fdesc(event), event->rd_kafka_headers, kafka->rd_kafka_handle, kafka->topic, rd_kafka_err2str(err));
            return FALSE;
        }
    } else {
        if (rd_kafka_produce(kafka->rd_kafka_topic, RD_KAFKA_PARTITION_UA, RD_KAFKA_MSG_F_COPY, event_string, event_length, NULL, 0, event) == -1) {
            int error = errno;
            event->status = N_KAFKA_EVENT_ERROR;
            event->error_time = time(NULL);
            n_log(LOG_ERR, "failed to produce event: %p%s, producer: %p, topic: %s, error: %s", event, n_kafka_event_fdesc(event), kafka->rd_kafka_handle, kafka->topic, strerror(error));
            return FALSE;
        }
    }
    // Success to *enqueue* event
    n_log(LOG_DEBUG, "successfully enqueued event %p%s in local producer %p : %s", event, n_kafka_event_fdesc(event), kafka->rd_kafka_handle, kafka->topic);

    event->status = N_KAFKA_EVENT_WAITING_ACK;

    return TRUE;
} /* n_kafka_produce_ex */

N_KAFKA_EVENT* n_kafka_new_event_from_char(const char* string, size_t written, int schema_id) {
    __n_assert(string, return NULL);

    size_t offset = 0;
    if (schema_id != -1)
        offset = 5;

    N_KAFKA_EVENT* event = n_kafka_new_event(schema_id);
    __n_assert(event, return NULL);

    Malloc(event->event_string, N_STR, 1);
    __n_assert(event->event_string, free(event); return NULL);

    // allocate the size of the event + (the size of the schema id + magic byte) + one ending \0
    size_t length = written + offset;
    Malloc(event->event_string->data, char, length);
    __n_assert(event->event_string->data, free(event->event_string); free(event); return NULL);
    event->event_string->length = length;

    // copy incomming body
    memcpy(event->event_string->data + offset, string, written);
    event->event_string->written = written + offset;

    if (schema_id != -1)
        n_kafka_put_schema_in_nstr(event->event_string, schema_id);

    // set status and schema id
    event->status = N_KAFKA_EVENT_QUEUED;

    return event;
} /* n_kafka_new_event_from_char */

N_KAFKA_EVENT* n_kafka_new_event_from_string(const N_STR* string, int schema_id) {
    __n_assert(string, return NULL);
    __n_assert(string->data, return NULL);

    N_KAFKA_EVENT* event = n_kafka_new_event_from_char(string->data, string->written, schema_id);

    return event;
} /* n_kafka_new_event_from_string */

N_KAFKA_EVENT* n_kafka_new_event_from_file(char* filename, int schema_id) {
    __n_assert(filename, return NULL);

    N_STR* from = file_to_nstr(filename);
    __n_assert(from, return NULL);

    N_KAFKA_EVENT* event = n_kafka_new_event_from_string(from, schema_id);
    free_nstr(&from);

    return event;
} /* n_kafka_new_event_from_file */

void n_kafka_event_destroy_ptr(void* event_ptr) {
    __n_assert(event_ptr, return);
    N_KAFKA_EVENT* event = (N_KAFKA_EVENT*)event_ptr;
    __n_assert(event, return);

    if (event->event_string)
        free_nstr(&event->event_string);

    if (event->event_files_to_delete)
        free_nstr(&event->event_files_to_delete);

    FreeNoLog(event->from_topic);

    if (event->rd_kafka_headers)
        rd_kafka_headers_destroy(event->rd_kafka_headers);

    if (event->received_headers)
        list_destroy(&event->received_headers);

    free(event);
    return;
} /* n_kafka_event_destroy_ptr */

int n_kafka_event_destroy(N_KAFKA_EVENT** event) {
    __n_assert(event && (*event), return FALSE);
    n_kafka_event_destroy_ptr((*event));
    (*event) = NULL;
    return TRUE;
} /* n_kafka_event_destroy */

int n_kafka_poll(N_KAFKA* kafka) {
    __n_assert(kafka, return FALSE);

    int event_consumption_enabled;
    int event_production_enabled;

    read_lock(kafka->rwlock);
    event_consumption_enabled = kafka->event_consumption_enabled;
    event_production_enabled = kafka->event_production_enabled;
    unlock(kafka->rwlock);

    // wait poll interval msecs for kafka response
    if (kafka->mode == RD_KAFKA_PRODUCER) {
        if (event_production_enabled) {
            int nb_events = rd_kafka_poll(kafka->rd_kafka_handle, kafka->poll_interval);
            (void)nb_events;
            /* transactional mode: a kafka transaction is opened lazily before the
             * first queued event is produced in this cycle, and committed after the
             * locked pass below. commit/abort must run outside the rwlock because
             * they synchronously fire the delivery callbacks which take that lock. */
            int txn_active = 0;
            int txn_begin_failed = 0;
            write_lock(kafka->rwlock);
            // check events status in event table
            LIST_NODE* node = kafka->events_to_send->start;
            while (node) {
                N_KAFKA_EVENT* event = (N_KAFKA_EVENT*)node->ptr;
                if (event->status == N_KAFKA_EVENT_OK) {
                    kafka->nb_waiting--;
                    n_log(LOG_DEBUG, "removing event OK %p%s", (void*)event, n_kafka_event_fdesc(event));
                    LIST_NODE* node_to_kill = node;
                    node = node->next;
                    N_KAFKA_EVENT* event_to_kill = remove_list_node(kafka->events_to_send, node_to_kill, N_KAFKA_EVENT);
                    n_kafka_event_destroy(&event_to_kill);
                    continue;
                } else if (event->status == N_KAFKA_EVENT_QUEUED) {
                    /* in transactional mode, open a transaction before the first produce */
                    if (kafka->is_transactional && !txn_active && !txn_begin_failed) {
                        rd_kafka_error_t* txn_err = rd_kafka_begin_transaction(kafka->rd_kafka_handle);
                        if (txn_err) {
                            n_log(LOG_ERR, "could not begin transaction on producer %p, topic %s: %s%s", kafka->rd_kafka_handle, kafka->topic, rd_kafka_error_string(txn_err), rd_kafka_error_is_fatal(txn_err) ? " (FATAL)" : "");
                            rd_kafka_error_destroy(txn_err);
                            /* leave the queued events untouched, retry on the next poll */
                            txn_begin_failed = 1;
                        } else {
                            txn_active = 1;
                        }
                    }
                    /* if a transaction was required but could not be opened, skip producing this cycle */
                    if (kafka->is_transactional && !txn_active) {
                        node = node->next;
                        continue;
                    }
                    if (n_kafka_produce_ex(kafka, event) == FALSE) {
                        /* produce_ex already set status to ERROR,
                         * adjust counters: QUEUED -> ERROR */
                        kafka->nb_queued--;
                        kafka->nb_error++;
                    } else {
                        kafka->nb_waiting++;
                        kafka->nb_queued--;
                    }
                } else if (event->status == N_KAFKA_EVENT_ERROR) {
                    /* counters already adjusted at transition time
                     * (in produce_ex failure above or in delivery callback) */
                    if (kafka->error_timeout > 0 &&
                        (time(NULL) - event->error_time) >= (time_t)kafka->error_timeout) {
                        n_log(LOG_INFO, "retrying errored event %p%s after %d s", (void*)event, n_kafka_event_fdesc(event), kafka->error_timeout);
                        event->status = N_KAFKA_EVENT_QUEUED;
                        event->error_time = 0;
                        kafka->nb_error--;
                        kafka->nb_queued++;
                    }
                }
                node = node->next;
            }
            unlock(kafka->rwlock);

            /* close the transaction outside the lock: commit_transaction (and
             * abort_transaction) drive the delivery callbacks that, on success,
             * move the produced files to 'sended' and, on failure, flip events
             * back to ERROR. Both callbacks take the rwlock themselves.
             * Retry the commit on retriable errors; on any other failure (or once
             * retries are exhausted) abort to bring the producer back to a clean
             * state: the purged messages are reported to the delivery callback as
             * errors so their files stay in 'to-send' and are produced again later. */
            if (txn_active) {
                int committed = 0;
                int attempts = 0;
                while (!committed && attempts < 5) {
                    attempts++;
                    rd_kafka_error_t* commit_err = rd_kafka_commit_transaction(kafka->rd_kafka_handle, -1);
                    if (!commit_err) {
                        committed = 1;
                        break;
                    }
                    n_log(LOG_ERR, "could not commit transaction (attempt %d) on producer %p, topic %s: %s", attempts, kafka->rd_kafka_handle, kafka->topic, rd_kafka_error_string(commit_err));
                    int retriable = rd_kafka_error_is_retriable(commit_err);
                    int fatal = rd_kafka_error_is_fatal(commit_err);
                    rd_kafka_error_destroy(commit_err);
                    if (retriable)
                        continue;  // resume the in-flight commit
                    if (fatal)
                        n_log(LOG_ERR, "FATAL transaction error on producer %p, topic %s, producer is no longer usable", kafka->rd_kafka_handle, kafka->topic);
                    break;  // abort-required, fatal or other non-retriable error
                }
                if (!committed) {
                    n_log(LOG_ERR, "aborting transaction on producer %p, topic %s", kafka->rd_kafka_handle, kafka->topic);
                    rd_kafka_error_t* abort_err = rd_kafka_abort_transaction(kafka->rd_kafka_handle, -1);
                    if (abort_err) {
                        n_log(LOG_ERR, "could not abort transaction on producer %p, topic %s: %s", kafka->rd_kafka_handle, kafka->topic, rd_kafka_error_string(abort_err));
                        rd_kafka_error_destroy(abort_err);
                    }
                }
            }
        } else {
            uint64_t sleep_us = (kafka->poll_interval > 0) ? ((uint64_t)(uint32_t)kafka->poll_interval * 1000ULL) : 0ULL;
            usleep((unsigned int)(sleep_us > (uint64_t)UINT_MAX ? (uint64_t)UINT_MAX : sleep_us));  // production is disabled, sleep instead of producing
        }
    } else if (kafka->mode == RD_KAFKA_CONSUMER) {
        if (event_consumption_enabled) {
            rd_kafka_message_t* rkm = NULL;
            while (event_consumption_enabled && (rkm = rd_kafka_consumer_poll(kafka->rd_kafka_handle, kafka->poll_interval))) {
                read_lock(kafka->rwlock);
                event_consumption_enabled = kafka->event_consumption_enabled;
                unlock(kafka->rwlock);
                if (rkm->err) {
                    /* Consumer errors are generally to be considered
                     * informational as the consumer will automatically
                     * try to recover from all types of errors. */
                    n_log(LOG_ERR, "consumer: %s", rd_kafka_message_errstr(rkm));
                    rd_kafka_message_destroy(rkm);
                    uint64_t sleep_us = (kafka->poll_interval > 0) ? ((uint64_t)(uint32_t)kafka->poll_interval * 1000ULL) : 0ULL;
                    usleep((unsigned int)(sleep_us > (uint64_t)UINT_MAX ? (uint64_t)UINT_MAX : sleep_us));
                    return FALSE;
                }
                // Reminder of rkm contents
                n_log(LOG_DEBUG, "Message on %s [%" PRId32 "] at offset %" PRId64 " (leader epoch %" PRId32 ")", rd_kafka_topic_name(rkm->rkt), rkm->partition, rkm->offset, rd_kafka_message_leader_epoch(rkm));

                // Print the message key
                if (rkm->key && rkm->key_len > 0)
                    n_log(LOG_DEBUG, "Key: %.*s", (int)rkm->key_len, (const char*)rkm->key);
                else if (rkm->key)
                    n_log(LOG_DEBUG, "Key: (%d bytes)", (int)rkm->key_len);

                if (rkm->payload && rkm->len > 0) {
                    write_lock(kafka->rwlock);
                    // make a copy of the event for further processing
                    N_KAFKA_EVENT* event = NULL;
                    event = n_kafka_new_event_from_char(rkm->payload, rkm->len, -1);  // no schema id because we want a full raw copy here
                    event->parent_table = kafka;
                    // test if there are headers, save them
                    rd_kafka_headers_t* hdrs = NULL;
                    if (!rd_kafka_message_headers(rkm, &hdrs)) {
                        size_t idx = 0;
                        const char* name = NULL;
                        const void* val = NULL;
                        size_t size = 0;
                        event->received_headers = new_generic_list(MAX_LIST_ITEMS);
                        while (!rd_kafka_header_get_all(hdrs, idx, &name, &val, &size)) {
                            N_STR* header_entry = NULL;
                            nstrprintf(header_entry, "%s=%s", _str(name), _str((char*)val));
                            list_push(event->received_headers, header_entry, &free_nstr_ptr);
                            idx++;
                        }
                    }
                    // save originating topic (can help sorting if there are multiples one)
                    event->from_topic = strdup(rd_kafka_topic_name(rkm->rkt));
                    if (kafka->schema_id != -1)
                        event->schema_id = n_kafka_get_schema_from_nstr(event->event_string);
                    list_push(kafka->received_events, event, &n_kafka_event_destroy_ptr);
                    n_log(LOG_DEBUG, "Consumer received event of (%d bytes) from topic %s", (int)rkm->len, event->from_topic);
                    unlock(kafka->rwlock);
                }
                rd_kafka_message_destroy(rkm);
            }
        } else {
            uint64_t sleep_us = (kafka->poll_interval > 0) ? ((uint64_t)(uint32_t)kafka->poll_interval * 1000ULL) : 0ULL;
            usleep((unsigned int)(sleep_us > (uint64_t)UINT_MAX ? (uint64_t)UINT_MAX : sleep_us));  // consumption is disabled, sleep instead of consuming
        }
    }
    // n_log( LOG_DEBUG , "kafka poll for handle %p returned %d elements" , kafka -> rd_kafka_handle , nb_events );
    return TRUE;
} /* n_kafka_poll */

void* n_kafka_polling_thread(void* ptr) {
    N_KAFKA* kafka = (N_KAFKA*)ptr;

    int status = 1;

    N_TIME chrono;

    if (kafka->mode == RD_KAFKA_PRODUCER)
        n_log(LOG_DEBUG, "starting polling thread for kafka handler %p mode PRODUCER (%d) topic %s", kafka->rd_kafka_handle, RD_KAFKA_PRODUCER, kafka->topic);
    if (kafka->mode == RD_KAFKA_CONSUMER) {
        char* topiclist = join(kafka->topics, ",");
        n_log(LOG_DEBUG, "starting polling thread for kafka handler %p mode CONSUMER (%d) topic %s", kafka->rd_kafka_handle, RD_KAFKA_CONSUMER, _str(topiclist));
        FreeNoLog(topiclist);
    }

    start_HiTimer(&chrono);

    int64_t remaining_time = (int64_t)kafka->poll_timeout * 1000;
    while (status == 1) {
        if (n_kafka_poll(kafka) == FALSE) {
            if (kafka->mode == RD_KAFKA_PRODUCER) {
                n_log(LOG_ERR, "failed to poll kafka producer handle %p with topic %s", kafka->rd_kafka_handle, rd_kafka_topic_name(kafka->rd_kafka_topic));
            } else if (kafka->topics) {
                char* topiclist = join(kafka->topics, ",");
                n_log(LOG_ERR, "failed to poll kafka consumer handle %p with topic %s", kafka->rd_kafka_handle, _str(topiclist));
                FreeNoLog(topiclist);
            }
        }

        read_lock(kafka->rwlock);
        status = kafka->polling_thread_status;
        unlock(kafka->rwlock);

        if (status == 2)
            break;

        int64_t elapsed_time = get_usec(&chrono);
        if (kafka->poll_timeout != -1) {
            remaining_time -= elapsed_time;
            if (remaining_time < 0) {
                if (kafka->mode == RD_KAFKA_PRODUCER) {
                    n_log(LOG_DEBUG, "timeouted on kafka handle %p", kafka->rd_kafka_handle);
                } else if (kafka->mode == RD_KAFKA_CONSUMER) {
                    n_log(LOG_DEBUG, "timeouted on kafka handle %p", kafka->rd_kafka_handle);
                }
                break;
            }
        }
        // n_log( LOG_DEBUG , "remaining time: %d on kafka handle %p" , remaining_time , kafka -> rd_kafka_handle );
    }

    write_lock(kafka->rwlock);
    kafka->polling_thread_status = 0;
    unlock(kafka->rwlock);

    n_log(LOG_DEBUG, "exiting polling thread for kafka handler %p mode %s", kafka->rd_kafka_handle, (kafka->mode == RD_KAFKA_PRODUCER) ? "PRODUCER" : "CONSUMER");
    pthread_exit(NULL);
    return NULL;
} /* n_kafka_polling_thread */

int n_kafka_start_polling_thread(N_KAFKA* kafka) {
    __n_assert(kafka, return FALSE);

    read_lock(kafka->rwlock);
    int status = kafka->polling_thread_status;
    unlock(kafka->rwlock);

    if (status == 1) {
        n_log(LOG_ERR, "kafka polling thread already started for handle %p", kafka);
        return FALSE;
    }

    write_lock(kafka->rwlock);
    kafka->polling_thread_status = 1;

    if (pthread_create(&kafka->polling_thread, NULL, n_kafka_polling_thread, (void*)kafka) != 0) {
        n_log(LOG_ERR, "unable to create polling_thread for kafka handle %p", kafka);
        unlock(kafka->rwlock);
        return FALSE;
    }
    unlock(kafka->rwlock);

    n_log(LOG_DEBUG, "pthread_create sucess for kafka handle %p->%p", kafka, kafka->rd_kafka_handle);

    return TRUE;
} /* n_kafka_start_polling_thread */

int n_kafka_stop_polling_thread(N_KAFKA* kafka) {
    __n_assert(kafka, return FALSE);

    if (kafka->mode == RD_KAFKA_CONSUMER) {
        n_kafka_disable_event_consumption(kafka);
    }

    read_lock(kafka->rwlock);
    int polling_thread_status = kafka->polling_thread_status;
    unlock(kafka->rwlock);

    if (polling_thread_status == 0) {
        n_log(LOG_DEBUG, "kafka polling thread already stopped for handle %p", kafka);
        return FALSE;
    }
    if (polling_thread_status == 2) {
        n_log(LOG_DEBUG, "kafka polling ask for stop thread already done for handle %p", kafka);
        return FALSE;
    }

    write_lock(kafka->rwlock);
    kafka->polling_thread_status = 2;
    unlock(kafka->rwlock);

    if (kafka->rd_kafka_handle && kafka->mode == RD_KAFKA_CONSUMER) {
        rd_kafka_consumer_close(kafka->rd_kafka_handle);
    }

    struct timespec ts;
    clock_gettime(CLOCK_REALTIME, &ts);
    ts.tv_sec += 10;

    int rc = pthread_timedjoin_np(kafka->polling_thread, NULL, &ts);
    if (rc != 0) {
        n_log(LOG_ERR, "polling thread did not stop in 10s, force stop !");
        return FALSE;
    }

    return TRUE;
} /* n_kafka_stop_polling_thread */

int n_kafka_enable_event_consumption(N_KAFKA* kafka) {
    __n_assert(kafka, return FALSE);
    write_lock(kafka->rwlock);
    kafka->event_consumption_enabled = TRUE;
    unlock(kafka->rwlock);
    return TRUE;
} /* n_kafka_set_consumer_event_polling */

int n_kafka_disable_event_consumption(N_KAFKA* kafka) {
    __n_assert(kafka, return FALSE);
    write_lock(kafka->rwlock);
    kafka->event_consumption_enabled = FALSE;
    unlock(kafka->rwlock);
    return TRUE;
} /* n_kafka_set_consumer_event_polling */

int n_kafka_enable_event_production(N_KAFKA* kafka) {
    __n_assert(kafka, return FALSE);
    write_lock(kafka->rwlock);
    kafka->event_production_enabled = TRUE;
    unlock(kafka->rwlock);
    return TRUE;
} /* n_kafka_set_consumer_event_polling */

int n_kafka_disable_event_production(N_KAFKA* kafka) {
    __n_assert(kafka, return FALSE);
    write_lock(kafka->rwlock);
    kafka->event_production_enabled = FALSE;
    unlock(kafka->rwlock);
    return TRUE;
} /* n_kafka_set_consumer_event_polling */

int n_kafka_dump_unprocessed(N_KAFKA* kafka, char* directory) {
    __n_assert(kafka, return FALSE);
    __n_assert(directory, return FALSE);

    int status = 0;
    size_t nb_todump = 0;
    read_lock(kafka->rwlock);
    status = kafka->polling_thread_status;
    nb_todump = kafka->nb_queued + kafka->nb_waiting + kafka->nb_error;
    if (status != 0) {
        n_log(LOG_ERR, "kafka handle %p thread polling func is still running, aborting dump", kafka);
        unlock(kafka->rwlock);
        return FALSE;
    }
    if (nb_todump == 0) {
        n_log(LOG_DEBUG, "kafka handle %p: nothing to dump, all events processed correctly", kafka);
        unlock(kafka->rwlock);
        return TRUE;
    }

    /* use a plain struct to alias event data without allocating a data buffer
     * that would leak when overwritten */
    N_STR* dumpstr = NULL;
    Malloc(dumpstr, N_STR, 1);
    __n_assert(dumpstr, unlock(kafka->rwlock); return FALSE);
    list_foreach(node, kafka->events_to_send) {
        N_KAFKA_EVENT* event = node->ptr;
        if (event->status != N_KAFKA_EVENT_OK) {
            size_t offset = 0;
            if (event->schema_id != -1)
                offset = 5;

            N_STR* filename = NULL;
            nstrprintf(filename, "%s/%s+%p", directory, kafka->topic, event);
            n_log(LOG_DEBUG, "Dumping unprocessed event %p%s to %s", (void*)event, n_kafka_event_fdesc(event), _nstr(filename));
            // dump event here: alias the data pointer, do not own it
            dumpstr->data = event->event_string->data + offset;
            dumpstr->written = event->event_string->written - offset;
            dumpstr->length = event->event_string->length - offset;
            nstr_to_file(dumpstr, _nstr(filename));
            free_nstr(&filename);
        }
    }
    unlock(kafka->rwlock);
    /* clear aliased pointer before freeing the wrapper struct */
    dumpstr->data = NULL;
    free_nstr(&dumpstr);
    return TRUE;
} /* n_kafka_dump_unprocessed */

int n_kafka_load_unprocessed(N_KAFKA* kafka, const char* directory) {
    __n_assert(kafka, return FALSE);
    __n_assert(directory, return FALSE);

    write_lock(kafka->rwlock);
    /* load events from filename using base64decode( filename ) split( result , '+' )
       to get brokersname+topic to check against what's saved with the event
       and what's inside kafka's handle conf */
    unlock(kafka->rwlock);

    return TRUE;
} /* n_kafka_load_unprocessed */

N_KAFKA_EVENT* n_kafka_get_event(N_KAFKA* kafka) {
    __n_assert(kafka, return NULL);

    N_KAFKA_EVENT* event = NULL;

    write_lock(kafka->rwlock);
    if (kafka->received_events->start)
        event = remove_list_node(kafka->received_events, kafka->received_events->start, N_KAFKA_EVENT);
    unlock(kafka->rwlock);

    return event;
} /* n_kafka_get_event */