out_kafka: document OTLP partition tradeoff

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

Author: edsiper

plugins/out_kafka/kafka.c

@@ -48,6 +48,18 @@ struct otlp_logs_resource_partition {
     msgpack_sbuffer buffer;
 };
 
+static const char *default_logs_body_keys[] = {"log", "message"};
+
+static void init_otlp_logs_options(struct flb_opentelemetry_otlp_logs_options *options)
+{
+    memset(options, 0, sizeof(*options));
+    options->logs_require_otel_metadata = FLB_FALSE;
+    options->logs_body_keys = default_logs_body_keys;
+    options->logs_body_key_count = sizeof(default_logs_body_keys) /
+                                   sizeof(default_logs_body_keys[0]);
+    options->logs_body_key_attributes = FLB_FALSE;
+}
+
 void cb_kafka_msg(rd_kafka_t *rk, const rd_kafka_message_t *rkmessage,
                   void *opaque)
 {
@@ -555,7 +567,7 @@ static int produce_raw_payload_with_key_retry_control(const void *payload,
         message_key_len = 0;
     }
 
-    if (queue_full_retry_limit == 0 && allow_engine_retry == FLB_FALSE) {
+    if (queue_full_retry_limit <= 0 && allow_engine_retry == FLB_FALSE) {
         queue_full_retry_limit = FLB_KAFKA_PARTIAL_QUEUE_FULL_RETRIES;
     }
 
@@ -971,7 +983,6 @@ static int produce_partitioned_otlp_logs(struct flb_out_kafka *ctx,
     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;
@@ -1081,11 +1092,7 @@ static int produce_partitioned_otlp_logs(struct flb_out_kafka *ctx,
         goto cleanup;
     }
 
-    memset(&options, 0, sizeof(options));
-    options.logs_require_otel_metadata = FLB_FALSE;
-    options.logs_body_keys = default_logs_body_keys;
-    options.logs_body_key_count = 2;
-    options.logs_body_key_attributes = FLB_FALSE;
+    init_otlp_logs_options(&options);
 
     for (index = 0; index < partition_count; index++) {
         partition = &partitions[index];
@@ -1122,9 +1129,12 @@ static int produce_partitioned_otlp_logs(struct flb_out_kafka *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.
+         * Partitioned OTLP log sends are at-most-once after the first partition
+         * is accepted by librdkafka. If a later partition fails, return FLB_ERROR
+         * instead of FLB_RETRY so engine replay does not duplicate partitions
+         * already enqueued. Under sustained back-pressure this can partially
+         * deliver the original chunk; disable this option or reduce Kafka
+         * back-pressure when chunk-level retry durability is required.
          */
         ret = produce_raw_payload_with_key_retry_control(payload,
                                                          flb_sds_len(payload),
@@ -1162,7 +1172,6 @@ static int produce_otlp_json(struct flb_out_kafka *ctx,
     int result;
     flb_sds_t payload;
     struct flb_opentelemetry_otlp_logs_options options;
-    static const char *default_logs_body_keys[] = {"log", "message"};
 
     payload = NULL;
 
@@ -1173,11 +1182,7 @@ static int produce_otlp_json(struct flb_out_kafka *ctx,
                                                  FLB_KAFKA_FMT_OTLP_JSON);
         }
 
-        memset(&options, 0, sizeof(options));
-        options.logs_require_otel_metadata = FLB_FALSE;
-        options.logs_body_keys = default_logs_body_keys;
-        options.logs_body_key_count = 2;
-        options.logs_body_key_attributes = FLB_FALSE;
+        init_otlp_logs_options(&options);
 
         payload = flb_opentelemetry_logs_to_otlp_json(event_chunk->data,
                                                       event_chunk->size,
@@ -1224,7 +1229,6 @@ static int produce_otlp_proto(struct flb_out_kafka *ctx,
     struct ctrace *ctr;
     flb_sds_t payload;
     struct flb_opentelemetry_otlp_logs_options options;
-    static const char *default_logs_body_keys[] = {"log", "message"};
 
     if (event_chunk->type == FLB_EVENT_TYPE_LOGS) {
         if (ctx->otlp_logs_partition_by_resource == FLB_TRUE) {
@@ -1233,11 +1237,7 @@ static int produce_otlp_proto(struct flb_out_kafka *ctx,
                                                  FLB_KAFKA_FMT_OTLP_PROTO);
         }
 
-        memset(&options, 0, sizeof(options));
-        options.logs_require_otel_metadata = FLB_FALSE;
-        options.logs_body_keys = default_logs_body_keys;
-        options.logs_body_key_count = 2;
-        options.logs_body_key_attributes = FLB_FALSE;
+        init_otlp_logs_options(&options);
 
         payload = flb_opentelemetry_logs_to_otlp_proto(event_chunk->data,
                                                        event_chunk->size,
@@ -1428,7 +1428,12 @@ static struct flb_config_map config_map[] = {
     FLB_CONFIG_MAP_BOOL, "otlp_logs_partition_by_resource", "false",
     0, FLB_TRUE, offsetof(struct flb_out_kafka, otlp_logs_partition_by_resource),
     "When using format otlp_json or otlp_proto, split OTLP log payloads by "
-    "resource and use a hash of the resource attributes as the Kafka message key."
+    "resource and use a hash of the resource attributes as the Kafka message key. "
+    "This supersedes message_key and message_key_field for those chunks; logs "
+    "without resource information are unkeyed. After partial partition delivery, "
+    "later produce failures are not retried by the engine to avoid duplicates; "
+    "disable this option or reduce Kafka back-pressure for chunk-level retry "
+    "durability."
    },
    {
     FLB_CONFIG_MAP_STR, "message_key", (char *)NULL,