[pkg/kafka/configkafka] Add conn_idle_timeout for franz-go clients

.chloggen/kafkaexporter-connection-idle-timeout.yaml

@@ -0,0 +1,29 @@
+# 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: Add `connection_idle_timeout` configuration option to control idle connection timeout.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [45321]
+
+# (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: |
+  Defaults to 4 minutes. It may take up to 2x the configured time before a connection is actually closed.
+  This setting is applicable for franz-go, while ignored in sarama because it does not close idle connections.
+
+# 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: []

.chloggen/kafkareceiver-connection-idle-timeout.yaml

@@ -0,0 +1,29 @@
+# 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: receiver/kafka
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add `connection_idle_timeout` configuration option to control idle connection timeout.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [45321]
+
+# (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: |
+  Defaults to 4 minutes. It may take up to 2x the configured time before a connection is actually closed.
+  This setting is applicable for franz-go, while ignored in sarama because it does not close idle connections.
+
+# 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: []

exporter/kafkaexporter/README.md

@@ -34,6 +34,7 @@ The following settings can be optionally configured:
 - `protocol_version` (default = 2.1.0): Kafka protocol version.
 - `resolve_canonical_bootstrap_servers_only` (default = false): Whether to resolve then reverse-lookup broker IPs during startup.
 - `client_id` (default = "otel-collector"): The client ID to configure the Kafka client with. The client ID will be used for all produce requests.
+- `connection_idle_timeout` (default = `4m`): The time after which idle connections to Kafka brokers are closed. Note: It may take up to 2x the configured time before a connection is actually closed. This setting is applicable for franz-go, while ignored in sarama because it does not close idle connections.
 - `logs`
   - `topic` (default = otlp\_logs): The name of the Kafka topic to which logs will be exported.
   - `encoding` (default = otlp\_proto): The encoding for logs. See [Supported encodings](#supported-encodings).

internal/kafka/franz_client.go

@@ -257,6 +257,10 @@ func commonOpts(ctx context.Context, clientCfg configkafka.ClientConfig,
 	if clientCfg.Metadata.RefreshInterval > 0 {
 		opts = append(opts, kgo.MetadataMaxAge(clientCfg.Metadata.RefreshInterval))
 	}
+	// Configure connection idle timeout
+	if clientCfg.ConnectionIdleTimeout > 0 {
+		opts = append(opts, kgo.ConnIdleTimeout(clientCfg.ConnectionIdleTimeout))
+	}
 	// Configure the min/max protocol version if provided
 	if clientCfg.ProtocolVersion != "" {
 		keyVersions := make(map[string]any)

pkg/kafka/configkafka/config.go

@@ -76,14 +76,21 @@ type ClientConfig struct {
 	//
 	// NOTE: this is experimental and may be removed in a future release.
 	UseLeaderEpoch bool `mapstructure:"use_leader_epoch"`
+
+	// ConnectionIdleTimeout specifies the time after which idle connections are closed.
+	//
+	// Note: It may take up to 2x the configured time before a connection is actually closed.
+	// This setting is applicable for franz-go, while ignored in sarama because it does not close idle connections.
+	ConnectionIdleTimeout time.Duration `mapstructure:"connection_idle_timeout"`
 }
 
 func NewDefaultClientConfig() ClientConfig {
 	return ClientConfig{
-		Brokers:        []string{"localhost:9092"},
-		ClientID:       "otel-collector",
-		Metadata:       NewDefaultMetadataConfig(),
-		UseLeaderEpoch: true,
+		Brokers:               []string{"localhost:9092"},
+		ClientID:              "otel-collector",
+		Metadata:              NewDefaultMetadataConfig(),
+		UseLeaderEpoch:        true,
+		ConnectionIdleTimeout: 4 * time.Minute,
 	}
 }
 
@@ -96,6 +103,9 @@ func (c ClientConfig) Validate() error {
 			return fmt.Errorf("invalid protocol version: %w", err)
 		}
 	}
+	if c.ConnectionIdleTimeout <= 0 {
+		return fmt.Errorf("connection_idle_timeout (%s) must be positive", c.ConnectionIdleTimeout)
+	}
 	return nil
 }
 

pkg/kafka/configkafka/config_test.go

@@ -118,6 +118,13 @@ func TestClientConfig(t *testing.T) {
 				return cfg
 			}(),
 		},
+		"conn_idle_timeout": {
+			expected: func() ClientConfig {
+				cfg := NewDefaultClientConfig()
+				cfg.ConnIdleTimeout = 5 * time.Minute
+				return cfg
+			}(),
+		},
 
 		// Invalid configurations
 		"brokers_required": {

pkg/kafka/configkafka/testdata/client_config.yaml

@@ -98,3 +98,6 @@ kafka/foo:
 
 kafka/not_use_leader_epoch:
   use_leader_epoch: false
+
+kafka/connection_idle_timeout:
+  connection_idle_timeout: 5m

receiver/kafkareceiver/README.md

@@ -76,6 +76,7 @@ The following settings can be optionally configured:
 - `client_id` (default = otel-collector): The consumer client ID that receiver will use
 - `rack_id` (default = ""): The rack identifier for this client. When set and brokers are configured with a rack-aware replica selector, the client will prefer fetching from the closest replica.
 - `use_leader_epoch` (default = true): (Experimental) When enabled, the consumer uses the leader epoch returned by brokers (KIP-320) to detect log truncation. Setting this to false clears the leader epoch from fetch offsets, disabling KIP-320. Disabling can improve compatibility with brokers that don’t fully support leader epochs (e.g., Azure Event Hubs), at the cost of losing automatic log-truncation safety.
+- `connection_idle_timeout` (default = `4m`): The time after which idle connections to Kafka brokers are closed. Note: It may take up to 2x the configured time before a connection is actually closed. This setting is applicable for franz-go, while ignored in sarama because it does not close idle connections.
 - `initial_offset` (default = latest): The initial offset to use if no offset was previously committed. Must be `latest` or `earliest`.
 - `session_timeout` (default = `10s`): The request timeout for detecting client failures when using Kafka’s group management facilities.
 - `heartbeat_interval` (default = `3s`): The expected time between heartbeats to the consumer coordinator when using Kafka’s group management facilities.

receiver/kafkareceiver/config_test.go

@@ -276,6 +276,36 @@ func TestLoadConfig(t *testing.T) {
 				},
 			},
 		},
+		{
+			id: component.NewIDWithName(metadata.Type, "connection_idle_timeout"),
+			expected: &Config{
+				ClientConfig: func() configkafka.ClientConfig {
+					config := configkafka.NewDefaultClientConfig()
+					config.ConnectionIdleTimeout = 5 * time.Minute
+					return config
+				}(),
+				ConsumerConfig: configkafka.NewDefaultConsumerConfig(),
+				Logs: TopicEncodingConfig{
+					Topics:   []string{"otlp_logs"},
+					Encoding: "otlp_proto",
+				},
+				Metrics: TopicEncodingConfig{
+					Topics:   []string{"otlp_metrics"},
+					Encoding: "otlp_proto",
+				},
+				Traces: TopicEncodingConfig{
+					Topics:   []string{"otlp_spans"},
+					Encoding: "otlp_proto",
+				},
+				Profiles: TopicEncodingConfig{
+					Topics:   []string{"otlp_profiles"},
+					Encoding: "otlp_proto",
+				},
+				ErrorBackOff: configretry.BackOffConfig{
+					Enabled: false,
+				},
+			},
+		},
 	}
 
 	for _, tt := range tests {

receiver/kafkareceiver/testdata/config.yaml

@@ -81,3 +81,6 @@ kafka/regex_topic_with_exclusion:
     exclude_topics:
     - "^traces-debug-.*$"
     encoding: otlp_proto
+
+kafka/connection_idle_timeout:
+  connection_idle_timeout: 5m