From 556c6901c4ca312bd9f0f6ca0f92b1e588e1da3b Mon Sep 17 00:00:00 2001 From: Pierre-Yves Ritschard Date: Tue, 3 May 2016 09:59:22 +0200 Subject: [PATCH] write_kafka: reliability when no key provided --- src/write_kafka.c | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/write_kafka.c b/src/write_kafka.c index 37b3e14f..10ae5a57 100644 --- a/src/write_kafka.c +++ b/src/write_kafka.c @@ -202,6 +202,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, @@ -252,6 +254,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); @@ -315,6 +318,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) -- 2.11.0