out_kafka: new option 'queue_full_retries' (#2553)

When the librdkafka queue is full and Fluent Bit cannot ingest more data,
it does a 'local retry' of maximum 10 times waiting a second between each
retry. But in some cases like the exposed in #2553 is not enough.

This patch exposes a new configuration property called 'queue_full_retries'
that configure the maximum number of times the retry must be done.

Note that this limit now can be disabled setting a value of '0' or 'false'.

Signed-off-by: Eduardo Silva <eduardo@treasure-data.com>

Author: edsiper

plugins/out_kafka/kafka.c

@@ -266,7 +266,14 @@ int produce_message(struct flb_time *tm, msgpack_object *map,
     }
 
  retry:
-    if (queue_full_retries >= 10) {
+    /*
+     * If the local rdkafka queue is full, we retry up to 'queue_full_retries'
+     * times set by the configuration (default: 10). If the configuration
+     * property was set to '0' or 'false', we don't impose a limit. Use that
+     * value under your own risk.
+     */
+    if (ctx->queue_full_retries > 0 &&
+        queue_full_retries >= ctx->queue_full_retries) {
         if (ctx->format == FLB_KAFKA_FMT_JSON) {
             flb_free(out_buf);
         }
@@ -294,8 +301,8 @@ int produce_message(struct flb_time *tm, msgpack_object *map,
          * otherwise let the caller to issue a main retry againt the engine.
          */
         if (rd_kafka_last_error() == RD_KAFKA_RESP_ERR__QUEUE_FULL) {
-            flb_plg_warn(ctx->ins, "internal queue is full, "
-                         "retrying in one second");
+            flb_plg_warn(ctx->ins,
+                         "internal queue is full, retrying in one second");
 
             /*
              * If the queue is full, first make sure to discard any further

plugins/out_kafka/kafka_config.c

@@ -186,6 +186,19 @@ struct flb_kafka *flb_kafka_conf_create(struct flb_output_instance *ins,
         }
     }
 
+    /* Config: queue_full_retries */
+    tmp = flb_output_get_property("queue_full_retries", ins);
+    if (!tmp) {
+        ctx->queue_full_retries = FLB_KAFKA_QUEUE_FULL_RETRIES;
+    }
+    else {
+        /* set number of retries: note that if the number is zero, means forever */
+        ctx->queue_full_retries = atoi(tmp);
+        if (ctx->queue_full_retries < 0) {
+            ctx->queue_full_retries = 0;
+        }
+    }
+
     /* Config Gelf_Timestamp_Key */
     tmp = flb_output_get_property("gelf_timestamp_key", ins);
     if (tmp) {

plugins/out_kafka/kafka_config.h

@@ -26,12 +26,13 @@
 
 #include "rdkafka.h"
 
-#define FLB_KAFKA_FMT_JSON  0
-#define FLB_KAFKA_FMT_MSGP  1
-#define FLB_KAFKA_FMT_GELF  2
-#define FLB_KAFKA_BROKERS   "127.0.0.1"
-#define FLB_KAFKA_TOPIC     "fluent-bit"
-#define FLB_KAFKA_TS_KEY    "@timestamp"
+#define FLB_KAFKA_FMT_JSON            0
+#define FLB_KAFKA_FMT_MSGP            1
+#define FLB_KAFKA_FMT_GELF            2
+#define FLB_KAFKA_BROKERS             "127.0.0.1"
+#define FLB_KAFKA_TOPIC               "fluent-bit"
+#define FLB_KAFKA_TS_KEY              "@timestamp"
+#define FLB_KAFKA_QUEUE_FULL_RETRIES  10
 
 /* rdkafka log levels based on syslog(3) */
 #define FLB_KAFKA_LOG_EMERG   0
@@ -93,6 +94,8 @@ struct flb_kafka {
 
     int dynamic_topic;
 
+    int queue_full_retries;
+
     /* Internal */
     rd_kafka_t *producer;
     rd_kafka_conf_t *conf;