Handle empty zap.Field{} to avoid panics

kafka/common.go

@@ -37,11 +37,13 @@ import (
 	"go.opentelemetry.io/otel/metric"
 	"go.opentelemetry.io/otel/trace"
 	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
 )
 
 // SASLMechanism type alias to sasl.Mechanism
 type SASLMechanism = sasl.Mechanism
 
+// TopicLogFieldFunc is a function that returns a zap.Field for a given topic.
 type TopicLogFieldFunc func(topic string) zap.Field
 
 // CommonConfig defines common configuration for Kafka consumers, producers,
@@ -240,6 +242,11 @@ func (cfg *CommonConfig) finalize() error {
 			}
 		}
 	}
+	// Wrap the cfg.TopicLogFieldFunc to ensure it never returns a field with
+	// an unknown type (like `zap.Field{}`).
+	if cfg.TopicLogFieldFunc != nil {
+		cfg.TopicLogFieldFunc = topicFieldFunc(cfg.TopicLogFieldFunc)
+	}
 	return errors.Join(errs...)
 }
 
@@ -329,3 +336,15 @@ func newAWSMSKIAMSASL() (sasl.Mechanism, error) {
 		}, nil
 	}), nil
 }
+
+func topicFieldFunc(f TopicLogFieldFunc) TopicLogFieldFunc {
+	return func(t string) zap.Field {
+		if f == nil {
+			return zap.Skip()
+		}
+		if field := f(t); field.Type > zapcore.UnknownType {
+			return field
+		}
+		return zap.Skip()
+	}
+}

kafka/common_test.go

@@ -47,6 +47,7 @@ func TestCommonConfig(t *testing.T) {
 		t.Helper()
 		err := in.finalize()
 		require.NoError(t, err)
+		in.TopicLogFieldFunc = nil
 		in.hooks = nil
 		assert.Equal(t, expected, in)
 	}
@@ -73,16 +74,13 @@ func TestCommonConfig(t *testing.T) {
 	})
 
 	t.Run("valid", func(t *testing.T) {
-		cfg := CommonConfig{
-			Brokers: []string{"broker"},
-			Logger:  zap.NewNop(),
-		}
-		err := cfg.finalize()
-		assert.NoError(t, err)
-		assert.Equal(t, CommonConfig{
+		assertValid(t, CommonConfig{
 			Brokers: []string{"broker"},
 			Logger:  zap.NewNop().Named("kafka"),
-		}, cfg)
+		}, CommonConfig{
+			Brokers: []string{"broker"},
+			Logger:  zap.NewNop(),
+		})
 	})
 
 	t.Run("brokers_from_environment", func(t *testing.T) {
@@ -300,3 +298,22 @@ func newClusterWithTopics(t testing.TB, partitions int32, topics ...string) (*kg
 
 	return client, addrs
 }
+
+func TestTopicFieldFunc(t *testing.T) {
+	t.Run("nil func", func(t *testing.T) {
+		topic := topicFieldFunc(nil)("a")
+		assert.Equal(t, zap.Skip(), topic)
+	})
+	t.Run("empty field", func(t *testing.T) {
+		topic := topicFieldFunc(func(topic string) zap.Field {
+			return zap.Field{}
+		})("b")
+		assert.Equal(t, zap.Skip(), topic)
+	})
+	t.Run("actual topic field", func(t *testing.T) {
+		topic := topicFieldFunc(func(topic string) zap.Field {
+			return zap.String("topic", topic)
+		})("c")
+		assert.Equal(t, zap.String("topic", "c"), topic)
+	})
+}

kafka/consumer_test.go

@@ -53,7 +53,7 @@ func TestNewConsumer(t *testing.T) {
 			cfg: ConsumerConfig{
 				CommonConfig: CommonConfig{
 					Brokers: []string{"localhost:invalidport"},
-					Logger:  zap.NewNop(),
+					Logger:  zapTest(t),
 				},
 				Topics:    []apmqueue.Topic{"topic"},
 				GroupID:   "groupid",
@@ -65,7 +65,7 @@ func TestNewConsumer(t *testing.T) {
 			cfg: ConsumerConfig{
 				CommonConfig: CommonConfig{
 					Brokers:  []string{"localhost:9092"},
-					Logger:   zap.NewNop(),
+					Logger:   zapTest(t),
 					ClientID: "clientid",
 					Version:  "1.0",
 					TLS:      &tls.Config{},
@@ -117,7 +117,7 @@ func TestConsumerHealth(t *testing.T) {
 			consumer := newConsumer(t, ConsumerConfig{
 				CommonConfig: CommonConfig{
 					Brokers: addrs,
-					Logger:  zap.NewNop(),
+					Logger:  zapTest(t),
 				},
 				Topics:    []apmqueue.Topic{"topic"},
 				GroupID:   "groupid",
@@ -155,7 +155,7 @@ func TestConsumerInstrumentation(t *testing.T) {
 	cfg := ConsumerConfig{
 		CommonConfig: CommonConfig{
 			Brokers:        addrs,
-			Logger:         zap.NewNop(),
+			Logger:         zapTest(t),
 			Namespace:      namespace,
 			TracerProvider: tp,
 		},
@@ -514,7 +514,7 @@ func TestConsumerContextPropagation(t *testing.T) {
 	addrs := newClusterAddrWithTopics(t, 2, "topic")
 	commonCfg := CommonConfig{
 		Brokers: addrs,
-		Logger:  zap.NewNop(),
+		Logger:  zapTest(t),
 	}
 	processed := make(chan struct{})
 	expectedMeta := map[string]string{
@@ -573,7 +573,7 @@ func TestMultipleConsumers(t *testing.T) {
 	cfg := ConsumerConfig{
 		CommonConfig: CommonConfig{
 			Brokers:   addrs,
-			Logger:    zap.NewNop(),
+			Logger:    zapTest(t),
 			Namespace: "name_space",
 		},
 		Topics:  []apmqueue.Topic{"topic"},
@@ -614,7 +614,7 @@ func TestMultipleConsumerGroups(t *testing.T) {
 	cfg := ConsumerConfig{
 		CommonConfig: CommonConfig{
 			Brokers:   addrs,
-			Logger:    zap.NewNop(),
+			Logger:    zapTest(t),
 			Namespace: "name_space",
 		},
 		Topics: []apmqueue.Topic{"topic"},
@@ -711,6 +711,58 @@ func TestConsumerRunError(t *testing.T) {
 	})
 }
 
+func TestConsumerTopicLogFieldFunc(t *testing.T) {
+	t.Run("empty field", func(t *testing.T) {
+		const partitions = 2
+		topic := apmqueue.Topic("topic")
+		event := apmqueue.Record{Topic: topic, Value: []byte("x")}
+		var counter atomic.Int64
+		client, addrs := newClusterWithTopics(t, partitions, string(topic))
+		cfg := ConsumerConfig{
+			CommonConfig: CommonConfig{
+				Brokers: addrs,
+				Logger:  zapTest(t),
+				TopicLogFieldFunc: func(topic string) zapcore.Field {
+					return zap.Field{}
+				},
+			},
+			GroupID: t.Name(),
+			Topics:  []apmqueue.Topic{topic},
+			Processor: apmqueue.ProcessorFunc(func(_ context.Context, r apmqueue.Record) error {
+				counter.Add(1)
+				assert.Equal(t, event.Topic, r.Topic)
+				assert.Equal(t, event.Value, r.Value)
+				// We cannot know which partition the record will be polled from,
+				// but we can assert it's in the range of number of partitions.
+				assert.True(t, event.Partition < partitions-1)
+				return nil
+			}),
+		}
+
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+		consumer := newConsumer(t, cfg)
+		go func() {
+			assert.NoError(t, consumer.Run(ctx))
+		}()
+
+		produceRecords := 100
+		for i := 0; i < produceRecords; i++ {
+			client.Produce(ctx, &kgo.Record{
+				Topic: string(topic),
+				Value: event.Value,
+			}, func(r *kgo.Record, err error) {
+				assert.NoError(t, err)
+				assert.NotNil(t, r)
+			})
+		}
+
+		assert.Eventually(t, func() bool {
+			return counter.Load() == int64(produceRecords)
+		}, time.Second, 50*time.Millisecond)
+	})
+}
+
 func newConsumer(t testing.TB, cfg ConsumerConfig) *Consumer {
 	if cfg.MaxPollWait <= 0 {
 		// Lower MaxPollWait, ShutdownGracePeriod to speed up execution.