out_kafka: add support to force flush when shutdown. (fluent#5593)

When fluent-bit shutdown for out_kafka plugin with rdkafka.linger.ms and
rdkafka.batch.size and then we show the warn message and lose the message.

So, this patch fixes the issue of not flushing due to the rdkafka linger.ms
option.

The 'rd_kafka_flush' function is linger.ms time will be ignored for the
duration of the call, queued messages will be sent to the broker as soon as
possible.

Signed-off-by: jinyong.choi <[email protected]>

Author: jinyongchoi

plugins/out_kafka/kafka.c

@@ -495,10 +495,29 @@ static void cb_kafka_flush(struct flb_event_chunk *event_chunk,
     FLB_OUTPUT_RETURN(FLB_OK);
 }
 
+static void kafka_flush_force(struct flb_out_kafka *ctx,
+                              struct flb_config *config)
+{
+    int ret;
+
+    if (!ctx) {
+        return;
+    }
+
+    if (ctx->kafka.rk) {
+        ret = rd_kafka_flush(ctx->kafka.rk, config->grace * 1000);
+        if (ret != RD_KAFKA_RESP_ERR_NO_ERROR) {
+            flb_plg_warn(ctx->ins, "Failed to force flush: %s",
+                         rd_kafka_err2str(ret));
+        }
+    }
+}
+
 static int cb_kafka_exit(void *data, struct flb_config *config)
 {
     struct flb_out_kafka *ctx = data;
 
+    kafka_flush_force(ctx, config);
     flb_out_kafka_destroy(ctx);
     return 0;
 }