out_kafka: avoid duplicate OTLP partition retries

Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>

Author: edsiper

plugins/out_kafka/kafka.c

@@ -520,9 +520,12 @@ int produce_message(struct flb_time *tm, msgpack_object *map,
     return FLB_OK;
 }
 
-static int produce_raw_payload_with_key(const void *payload, size_t payload_size,
-                                        char *key, size_t key_len,
-                                        struct flb_out_kafka *ctx)
+static int produce_raw_payload_with_key_retry_control(const void *payload,
+                                                      size_t payload_size,
+                                                      char *key,
+                                                      size_t key_len,
+                                                      int allow_engine_retry,
+                                                      struct flb_out_kafka *ctx)
 {
     int ret;
     int queue_full_retries;
@@ -551,7 +554,8 @@ static int produce_raw_payload_with_key(const void *payload, size_t payload_size
     }
 
 retry:
-    if (ctx->queue_full_retries > 0 &&
+    if (allow_engine_retry == FLB_TRUE &&
+        ctx->queue_full_retries > 0 &&
         queue_full_retries >= ctx->queue_full_retries) {
         ctx->blocked = FLB_FALSE;
         return FLB_RETRY;
@@ -588,6 +592,18 @@ static int produce_raw_payload_with_key(const void *payload, size_t payload_size
     return FLB_OK;
 }
 
+static int produce_raw_payload_with_key(const void *payload, size_t payload_size,
+                                        char *key, size_t key_len,
+                                        struct flb_out_kafka *ctx)
+{
+    return produce_raw_payload_with_key_retry_control(payload,
+                                                      payload_size,
+                                                      key,
+                                                      key_len,
+                                                      FLB_TRUE,
+                                                      ctx);
+}
+
 static int produce_raw_payload(const void *payload, size_t payload_size,
                                struct flb_out_kafka *ctx)
 {
@@ -903,6 +919,7 @@ static int get_otlp_group_resource(msgpack_object *group_metadata,
     if (ret != 0) {
         return -1;
     }
+    (void) scope_id;
 
     return 0;
 }
@@ -931,11 +948,13 @@ static int produce_partitioned_otlp_logs(struct flb_out_kafka *ctx,
     struct otlp_logs_resource_partition *current_partition;
     struct otlp_logs_resource_partition *partitions;
     struct flb_opentelemetry_otlp_logs_options options;
+    size_t produced_count;
     static const char *default_logs_body_keys[] = {"log", "message"};
 
     partitions = NULL;
     partition_count = 0;
     current_partition = NULL;
+    produced_count = 0;
 
     ret = flb_log_event_decoder_init(&decoder,
                                      (char *) event_chunk->data,
@@ -1080,11 +1099,17 @@ static int produce_partitioned_otlp_logs(struct flb_out_kafka *ctx,
             key_len = 0;
         }
 
-        ret = produce_raw_payload_with_key(payload,
-                                           flb_sds_len(payload),
-                                           key,
-                                           key_len,
-                                           ctx);
+        /*
+         * Once a partition has been accepted by librdkafka, returning FLB_RETRY
+         * for a later queue-full condition would replay the original chunk and
+         * duplicate the partitions already enqueued.
+         */
+        ret = produce_raw_payload_with_key_retry_control(payload,
+                                                         flb_sds_len(payload),
+                                                         key,
+                                                         key_len,
+                                                         produced_count == 0,
+                                                         ctx);
 
         if (format == FLB_KAFKA_FMT_OTLP_JSON) {
             flb_sds_destroy(payload);
@@ -1096,6 +1121,7 @@ static int produce_partitioned_otlp_logs(struct flb_out_kafka *ctx,
         if (ret != FLB_OK) {
             goto cleanup;
         }
+        produced_count++;
     }
 
     ret = FLB_OK;