kafkareceiver: refactor to make consumer generic

Remove code that is repeated for every signal,
and implement a generic consumer that calls out
to a signal-specific message handler.

We also update all the tests to use kfake rather
than mocks so we're exercising all of the receiver.
These tests are parallelised, so all tests run
faster than before, and with greater coverage.

Finally, we add `max_fetch_wait` (defaults to 250ms)
to the receiver's configuration. This was added for
speeding up the tests, but may also be useful for
users of the receiver.

Author: axw

.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/go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/credentials v1.17.66
 	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475
 	github.com/stretchr/testify v1.10.0
+	github.com/twmb/franz-go/pkg/kadm v1.15.0
 	github.com/twmb/franz-go/pkg/kfake v0.0.0-20250320172111-35ab5e5f5327
 	github.com/xdg-go/scram v1.1.2
 	go.opentelemetry.io/collector/component v1.29.1-0.20250411074447-4fb7c24ebecc

internal/kafka/go.sum

@@ -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
 }

internal/kafka/kafkatest/cluster.go

@@ -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/README.md

@@ -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/factory_test.go

@@ -14,7 +14,10 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/azure v0.123.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/jaeger v0.123.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/zipkin v0.123.0
+	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/kfake v0.0.0-20250320172111-35ab5e5f5327
 	go.opentelemetry.io/collector/client v1.29.1-0.20250411074447-4fb7c24ebecc
 	go.opentelemetry.io/collector/component v1.29.1-0.20250411074447-4fb7c24ebecc
 	go.opentelemetry.io/collector/component/componenttest v0.123.1-0.20250411074447-4fb7c24ebecc
@@ -88,8 +91,9 @@ 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/kadm v1.15.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