kafkareceiver: refactor to make consumer generic

.chloggen/kafkareceiver-generic.yaml

@@ -0,0 +1,30 @@
+# 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. filelogreceiver)
+component: kafkareceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add `max_fetch_wait` config setting
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [39360]
+
+# (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: |
+  This setting allows you to specify the maximum time that the broker will wait for
+  min_fetch_size bytes of data to be available before sending a response to the client.
+  Defaults to 250ms.
+
+# 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]

internal/kafka/client.go

@@ -59,6 +59,7 @@ func NewSaramaConsumerGroup(
 	saramaConfig.Consumer.Fetch.Min = consumerConfig.MinFetchSize
 	saramaConfig.Consumer.Fetch.Default = consumerConfig.DefaultFetchSize
 	saramaConfig.Consumer.Fetch.Max = consumerConfig.MaxFetchSize
+	saramaConfig.Consumer.MaxWaitTime = consumerConfig.MaxFetchWait
 	saramaConfig.Consumer.Offsets.AutoCommit.Enable = consumerConfig.AutoCommit.Enable
 	saramaConfig.Consumer.Offsets.AutoCommit.Interval = consumerConfig.AutoCommit.Interval
 	saramaConfig.Consumer.Offsets.Initial = saramaInitialOffsets[consumerConfig.InitialOffset]

internal/kafka/configkafka/config.go

@@ -98,6 +98,10 @@ type ConsumerConfig struct {
 
 	// The maximum bytes per fetch from Kafka (default "0", no limit)
 	MaxFetchSize int32 `mapstructure:"max_fetch_size"`
+
+	// The maximum amount of time to wait for MinFetchSize bytes to be
+	// available before the broker returns a response (default 250ms)
+	MaxFetchWait time.Duration `mapstructure:"max_fetch_wait"`
 }
 
 func NewDefaultConsumerConfig() ConsumerConfig {
@@ -112,6 +116,7 @@ func NewDefaultConsumerConfig() ConsumerConfig {
 		},
 		MinFetchSize:     1,
 		MaxFetchSize:     0,
+		MaxFetchWait:     250 * time.Millisecond,
 		DefaultFetchSize: 1048576,
 	}
 }

internal/kafka/configkafka/config_test.go

@@ -141,6 +141,7 @@ func TestConsumerConfig(t *testing.T) {
 				MinFetchSize:     10,
 				DefaultFetchSize: 1024,
 				MaxFetchSize:     4096,
+				MaxFetchWait:     time.Second,
 			},
 		},
 

internal/kafka/configkafka/testdata/consumer_config.yaml

@@ -10,6 +10,7 @@ kafka/full:
   min_fetch_size: 10
   default_fetch_size: 1024
   max_fetch_size: 4096
+  max_fetch_wait: 1s
 
 # Invalid configurations
 kafka/invalid_initial_offset:

internal/kafka/kafkatest/cluster.go

@@ -21,5 +21,7 @@ func NewCluster(tb testing.TB, opts ...kfake.Opt) (*kfake.Cluster, configkafka.C
 
 	cfg := configkafka.NewDefaultClientConfig()
 	cfg.Brokers = cluster.ListenAddrs()
+	// We need to set the protocol version to 2.3.0 to make Sarama happy.
+	cfg.ProtocolVersion = "2.3.0"
 	return cluster, cfg
 }

receiver/kafkareceiver/README.md

@@ -47,6 +47,7 @@ The following settings can be optionally configured:
 - `min_fetch_size` (default = `1`): The minimum number of message bytes to fetch in a request, defaults to 1 byte.
 - `default_fetch_size` (default = `1048576`): The default number of message bytes to fetch in a request, defaults to 1MB.
 - `max_fetch_size` (default = `0`): The maximum number of message bytes to fetch in a request, defaults to unlimited.
+- `max_fetch_wait` (default = `250ms`): The maximum amount of time the broker should wait for `min_fetch_size` bytes to be available before returning anyway.
 - `tls`: see [TLS Configuration Settings](https://github.com/open-telemetry/opentelemetry-collector/blob/main/config/configtls/README.md) for the full set of available options.
 - `auth`
   - `plain_text` (Deprecated in v0.123.0: use sasl with mechanism set to PLAIN instead.)

receiver/kafkareceiver/factory_test.go

@@ -41,7 +41,7 @@ func TestWithTracesUnmarshalers(t *testing.T) {
 		cfg := createDefaultConfig()
 		cfg.Traces.Encoding = "custom"
 		receiver, err := f.CreateTraces(context.Background(), receivertest.NewNopSettings(metadata.Type), cfg, nil)
-		tracesConsumer, ok := receiver.(*kafkaTracesConsumer)
+		tracesConsumer, ok := receiver.(*kafkaConsumer)
 		require.True(t, ok)
 		require.Equal(t, "custom", tracesConsumer.config.Traces.Encoding)
 		require.NoError(t, err)
@@ -50,7 +50,7 @@ func TestWithTracesUnmarshalers(t *testing.T) {
 	t.Run("default_encoding", func(t *testing.T) {
 		cfg := createDefaultConfig()
 		receiver, err := f.CreateTraces(context.Background(), receivertest.NewNopSettings(metadata.Type), cfg, nil)
-		tracesConsumer, ok := receiver.(*kafkaTracesConsumer)
+		tracesConsumer, ok := receiver.(*kafkaConsumer)
 		require.True(t, ok)
 		require.Equal(t, defaultTracesEncoding, tracesConsumer.config.Traces.Encoding)
 		require.NoError(t, err)
@@ -75,7 +75,7 @@ func TestWithMetricsUnmarshalers(t *testing.T) {
 		cfg := createDefaultConfig()
 		cfg.Metrics.Encoding = "custom"
 		receiver, err := f.CreateMetrics(context.Background(), receivertest.NewNopSettings(metadata.Type), cfg, nil)
-		metricsConsumer, ok := receiver.(*kafkaMetricsConsumer)
+		metricsConsumer, ok := receiver.(*kafkaConsumer)
 		require.True(t, ok)
 		require.Equal(t, "custom", metricsConsumer.config.Metrics.Encoding)
 		require.NoError(t, err)
@@ -84,7 +84,7 @@ func TestWithMetricsUnmarshalers(t *testing.T) {
 	t.Run("default_encoding", func(t *testing.T) {
 		cfg := createDefaultConfig()
 		receiver, err := f.CreateMetrics(context.Background(), receivertest.NewNopSettings(metadata.Type), cfg, nil)
-		metricsConsumer, ok := receiver.(*kafkaMetricsConsumer)
+		metricsConsumer, ok := receiver.(*kafkaConsumer)
 		require.True(t, ok)
 		require.Equal(t, defaultMetricsEncoding, metricsConsumer.config.Metrics.Encoding)
 		require.NoError(t, err)
@@ -109,7 +109,7 @@ func TestWithLogsUnmarshalers(t *testing.T) {
 		cfg := createDefaultConfig()
 		cfg.Logs.Encoding = "custom"
 		receiver, err := f.CreateLogs(context.Background(), receivertest.NewNopSettings(metadata.Type), cfg, nil)
-		logsConsumer, ok := receiver.(*kafkaLogsConsumer)
+		logsConsumer, ok := receiver.(*kafkaConsumer)
 		require.True(t, ok)
 		require.Equal(t, "custom", logsConsumer.config.Logs.Encoding)
 		require.NoError(t, err)
@@ -118,7 +118,7 @@ func TestWithLogsUnmarshalers(t *testing.T) {
 	t.Run("default_encoding", func(t *testing.T) {
 		cfg := createDefaultConfig()
 		receiver, err := f.CreateLogs(context.Background(), receivertest.NewNopSettings(metadata.Type), cfg, nil)
-		logsConsumer, ok := receiver.(*kafkaLogsConsumer)
+		logsConsumer, ok := receiver.(*kafkaConsumer)
 		require.True(t, ok)
 		require.Equal(t, defaultLogsEncoding, logsConsumer.config.Logs.Encoding)
 		require.NoError(t, err)

receiver/kafkareceiver/go.mod

@@ -14,7 +14,11 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/azure v0.124.1
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/jaeger v0.124.1
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/zipkin v0.124.1
+	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475
 	github.com/stretchr/testify v1.10.0
+	github.com/twmb/franz-go v1.18.1
+	github.com/twmb/franz-go/pkg/kadm v1.16.0
+	github.com/twmb/franz-go/pkg/kfake v0.0.0-20250320172111-35ab5e5f5327
 	go.opentelemetry.io/collector/client v1.30.0
 	go.opentelemetry.io/collector/component v1.30.0
 	go.opentelemetry.io/collector/component/componenttest v0.124.0
@@ -88,8 +92,8 @@ require (
 	github.com/openzipkin/zipkin-go v0.4.3 // indirect
 	github.com/pierrec/lz4/v4 v4.1.22 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
-	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
 	github.com/relvacode/iso8601 v1.6.0 // indirect
+	github.com/twmb/franz-go/pkg/kmsg v1.9.0 // indirect
 	github.com/xdg-go/pbkdf2 v1.0.0 // indirect
 	github.com/xdg-go/scram v1.1.2 // indirect
 	github.com/xdg-go/stringprep v1.0.4 // indirect