out_kafka: fix broken config map(#5097)

Signed-off-by: Takahiro Yamashita <[email protected]>

Author: nokute78

plugins/out_kafka/kafka.c

@@ -541,8 +541,8 @@ static struct flb_config_map config_map[] = {
    },
    {
     FLB_CONFIG_MAP_INT, "queue_full_retries", FLB_KAFKA_QUEUE_FULL_RETRIES,
-    0, FLB_TRUE, offsetof(struct flb_out_kafka, timestamp_format_str),
-    "Set the format the timestamp is in."
+    0, FLB_TRUE, offsetof(struct flb_out_kafka, queue_full_retries),
+    "Set the number of local retries to enqueue the data."
    },
    {
     FLB_CONFIG_MAP_STR, "gelf_timestamp_key", (char *)NULL,

plugins/out_kafka/kafka_config.c

@@ -30,6 +30,7 @@
 struct flb_out_kafka *flb_out_kafka_create(struct flb_output_instance *ins,
                                            struct flb_config *config)
 {
+    int ret;
     const char *tmp;
     char errstr[512];
     struct mk_list *head;
@@ -46,6 +47,13 @@ struct flb_out_kafka *flb_out_kafka_create(struct flb_output_instance *ins,
     ctx->ins = ins;
     ctx->blocked = FLB_FALSE;
 
+    ret = flb_output_config_map_set(ins, (void*) ctx);
+    if (ret == -1) {
+        flb_plg_error(ins, "unable to load configuration.");
+        flb_free(ctx);
+        return -1;
+    }
+
     /* rdkafka config context */
     ctx->conf = flb_kafka_conf_create(&ctx->kafka, &ins->properties, 0);
     if (!ctx->conf) {