out_kafka: use kv interface to query properties
#include <fluent-bit/flb_info.h>
#include <fluent-bit/flb_output.h>
#include <fluent-bit/flb_mem.h>
+#include <fluent-bit/flb_kv.h>
#include <fluent-bit/flb_utils.h>
#include "kafka_config.h"
@@ -37,7 +38,7 @@ struct flb_kafka *flb_kafka_conf_create(struct flb_output_instance *ins,
struct mk_list *topics;
struct flb_split_entry *entry;
struct flb_kafka *ctx;
- struct flb_config_prop *prop;
+ struct flb_kv *kv;
/* Configuration context */
ctx = flb_calloc(1, sizeof(struct flb_kafka));
@@ -84,15 +85,15 @@ struct flb_kafka *flb_kafka_conf_create(struct flb_output_instance *ins,
/* Iterate custom rdkafka properties */
mk_list_foreach(head, &ins->properties) {
- prop = mk_list_entry(head, struct flb_config_prop, _head);
- if (strncasecmp(prop->key, "rdkafka.", 8) == 0 &&
- strlen(prop->key) > 8) {
+ kv = mk_list_entry(head, struct flb_kv, _head);
+ if (strncasecmp(kv->key, "rdkafka.", 8) == 0 &&
+ flb_sds_len(kv->key) > 8) {
- ret = rd_kafka_conf_set(ctx->conf, prop->key + 8, prop->val,
+ ret = rd_kafka_conf_set(ctx->conf, kv->key + 8, kv->val,
errstr, sizeof(errstr));
if (ret != RD_KAFKA_CONF_OK) {
flb_error("[out_kafka] cannot configure '%s' property",
- prop->key + 8);
+ kv->key + 8);
}
}
}
