X-Git-Url: https://git.octo.it/?a=blobdiff_plain;f=src%2Fwrite_kafka.c;h=e881593b65161e20af6b697dd122ca2e402829c0;hb=edd9af8a874ebc9f2a7f02846807229a648917db;hp=2c3296db221eda6cdd2a8223bca1113950f9d5c5;hpb=5399e3b87299dae890d2820d2bcac05b7267a07b;p=collectd.git diff --git a/src/write_kafka.c b/src/write_kafka.c index 2c3296db..e881593b 100644 --- a/src/write_kafka.c +++ b/src/write_kafka.c @@ -36,7 +36,6 @@ #include #include -#include #include #include @@ -115,7 +114,6 @@ static int kafka_handle(struct kafka_topic_context *ctx) /* {{{ */ INFO ("write_kafka plugin: created KAFKA handle : %s", rd_kafka_name(ctx->kafka)); #if defined(HAVE_LIBRDKAFKA_LOGGER) && !defined(HAVE_LIBRDKAFKA_LOG_CB) - if rd_kafka_set_logger(ctx->kafka, kafka_log); #endif } @@ -203,6 +201,8 @@ static int kafka_write(const data_set_t *ds, /* {{{ */ key = ctx->key; if (key != NULL) keylen = strlen (key); + else + keylen = 0; rd_kafka_produce(ctx->topic, RD_KAFKA_PARTITION_UA, RD_KAFKA_MSG_F_COPY, buffer, blen, @@ -253,6 +253,7 @@ static void kafka_config_topic(rd_kafka_conf_t *conf, oconfig_item_t *ci) /* {{{ tctx->escape_char = '.'; tctx->store_rates = 1; tctx->format = KAFKA_FORMAT_JSON; + tctx->key = NULL; if ((tctx->kafka_conf = rd_kafka_conf_dup(conf)) == NULL) { sfree(tctx); @@ -316,6 +317,7 @@ static void kafka_config_topic(rd_kafka_conf_t *conf, oconfig_item_t *ci) /* {{{ } else if (strcasecmp ("Key", child->key) == 0) { cf_util_get_string (child, &tctx->key); + assert (tctx->key != NULL); } else if (strcasecmp ("Format", child->key) == 0) { status = cf_util_get_string(child, &key); if (status != 0)