kafkaexporter: add producer.linger config option (open-telemetry#44076)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description
Adds a new configuration option to the Kafka exporter to control the
linger time for the producer. Since `franz-go` now defaults to `10ms`,
it's best to allow users to configure this option to suit their needs.

<!-- Issue number (e.g. open-telemetry#1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
Fixes open-telemetry#44075

<!--Describe what testing was performed and which tests were added.-->
#### Testing

Unit tested

<!--Describe the documentation added.-->
#### Documentation

Docs addded.

---------

Signed-off-by: Marc Lopez Rubio <[email protected]>
Co-authored-by: Andrew Wilkins <[email protected]>

Author: marclop

.chloggen/f_add-kafka_exporter-producer.linger.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. receiver/filelog)
+component: exporter/kafka
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Adds a new configuration option to the Kafka exporter to control the linger time for the producer.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [44075]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext: Since `franz-go` now defaults to `10ms`, it's best to allow users to configure this option to suit their needs.
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

exporter/kafkaexporter/README.md

@@ -122,6 +122,7 @@ The following settings can be optionally configured:
         No compression levels supported yet
   - `flush_max_messages` (default = 0) The maximum number of messages the producer will send in a single broker request.
   - `allow_auto_topic_creation` (default = true) whether the broker is allowed to automatically create topics when they are referenced but do not already exist.
+  - `linger`: (default = `10ms`) How long individual topic partitions will linger waiting for more records before triggering a request to be built.
 
 ### Supported encodings
 

internal/kafka/franz_client.go

@@ -58,6 +58,7 @@ func NewFranzSyncProducer(ctx context.Context, clientCfg configkafka.ClientConfi
 		// the legacy compatibility sarama hashing to avoid hashing to different
 		// partitions in case partitioning is enabled.
 		kgo.RecordPartitioner(newSaramaCompatPartitioner()),
+		kgo.ProducerLinger(cfg.Linger),
 	)...)
 	if err != nil {
 		return nil, err

pkg/kafka/configkafka/config.go

@@ -219,6 +219,10 @@ type ProducerConfig struct {
 	// Whether or not to allow automatic topic creation.
 	// (default enabled).
 	AllowAutoTopicCreation bool `mapstructure:"allow_auto_topic_creation"`
+
+	// Linger controls the linger time for the producer.
+	// (default 10ms).
+	Linger time.Duration `mapstructure:"linger"`
 }
 
 func NewDefaultProducerConfig() ProducerConfig {
@@ -228,6 +232,7 @@ func NewDefaultProducerConfig() ProducerConfig {
 		Compression:            "none",
 		FlushMaxMessages:       0,
 		AllowAutoTopicCreation: true,
+		Linger:                 10 * time.Millisecond,
 	}
 }
 

pkg/kafka/configkafka/config_test.go

@@ -193,6 +193,7 @@ func TestProducerConfig(t *testing.T) {
 				},
 				FlushMaxMessages:       2,
 				AllowAutoTopicCreation: true,
+				Linger:                 10 * time.Millisecond,
 			},
 		},
 		"default_compression_level": {
@@ -206,6 +207,7 @@ func TestProducerConfig(t *testing.T) {
 				},
 				FlushMaxMessages:       2,
 				AllowAutoTopicCreation: true,
+				Linger:                 10 * time.Millisecond,
 			},
 		},
 		"snappy_compression": {
@@ -214,6 +216,7 @@ func TestProducerConfig(t *testing.T) {
 				RequiredAcks:           1,
 				Compression:            "snappy",
 				AllowAutoTopicCreation: true,
+				Linger:                 10 * time.Millisecond,
 			},
 		},
 		"disable_auto_topic_creation": {
@@ -222,6 +225,25 @@ func TestProducerConfig(t *testing.T) {
 				RequiredAcks:           1,
 				Compression:            "none",
 				AllowAutoTopicCreation: false,
+				Linger:                 10 * time.Millisecond,
+			},
+		},
+		"producer_linger": {
+			expected: ProducerConfig{
+				MaxMessageBytes:        1000000,
+				RequiredAcks:           1,
+				Compression:            "none",
+				AllowAutoTopicCreation: true,
+				Linger:                 100 * time.Millisecond,
+			},
+		},
+		"producer_linger_1s": {
+			expected: ProducerConfig{
+				MaxMessageBytes:        1000000,
+				RequiredAcks:           1,
+				Compression:            "none",
+				AllowAutoTopicCreation: true,
+				Linger:                 1 * time.Second,
 			},
 		},
 		"invalid_compression_level": {

pkg/kafka/configkafka/testdata/producer_config.yaml

@@ -31,3 +31,9 @@ kafka/invalid_compression:
   compression: brotli
 kafka/invalid_required_acks:
   required_acks: 3
+
+kafka/producer_linger:
+  linger: 100ms
+
+kafka/producer_linger_1s:
+  linger: 1s