Add support for go.delivery.report.fields

This configuration property will allow users to enable/disable certain fields of the delivery reports to reduce allocations.

Only the key & value are currently supported

Author: kevinconaway

kafka/handle.go

@@ -18,6 +18,7 @@ package kafka
 
 import (
 	"fmt"
+	"strings"
 	"sync"
 	"time"
 	"unsafe"
@@ -113,6 +114,9 @@ type handle struct {
 	// Forward delivery reports on Producer.Events channel
 	fwdDr bool
 
+	// Enabled fields for delivery reports
+	msgFields *messageFields
+
 	//
 	// consumer
 	//
@@ -134,6 +138,9 @@ func (h *handle) setup() {
 	h.rktNameCache = make(map[*C.rd_kafka_topic_t]string)
 	h.cgomap = make(map[int]cgoif)
 	h.name = C.GoString(C.rd_kafka_name(h.rk))
+	if h.msgFields == nil {
+		h.msgFields = newMessageFields()
+	}
 }
 
 func (h *handle) cleanup() {
@@ -323,3 +330,48 @@ func (h *handle) setOAuthBearerTokenFailure(errstr string) error {
 	}
 	return newError(cErr)
 }
+
+// messageFields controls which fields are made available for producer delivery reports & incoming messages
+// true values indicate that the field should be included
+type messageFields struct {
+	Key   bool
+	Value bool
+}
+
+// disableAll disable all fields
+func (mf *messageFields) disableAll() {
+	mf.Key = false
+	mf.Value = false
+}
+
+// newMessageFields returns a new messageFields with all fields enabled
+func newMessageFields() *messageFields {
+	return &messageFields{
+		Key:   true,
+		Value: true,
+	}
+}
+
+// newMessageFieldsFrom constructs a new messageFields from the given configuration value
+func newMessageFieldsFrom(v ConfigValue) (*messageFields, error) {
+	msgFields := newMessageFields()
+	switch v {
+	case "all":
+		// nothing to do
+	case "", "none":
+		msgFields.disableAll()
+	default:
+		msgFields.disableAll()
+		for _, value := range strings.Split(v.(string), ",") {
+			switch value {
+			case "key":
+				msgFields.Key = true
+			case "value":
+				msgFields.Value = true
+			default:
+				return nil, fmt.Errorf("unknown message field: %s", value)
+			}
+		}
+	}
+	return msgFields, nil
+}

kafka/message.go

@@ -135,10 +135,10 @@ func (h *handle) setupMessageFromC(msg *Message, cmsg *C.rd_kafka_message_t) {
 		msg.TopicPartition.Topic = &topic
 	}
 	msg.TopicPartition.Partition = int32(cmsg.partition)
-	if cmsg.payload != nil {
+	if cmsg.payload != nil && h.msgFields.Value {
 		msg.Value = C.GoBytes(unsafe.Pointer(cmsg.payload), C.int(cmsg.len))
 	}
-	if cmsg.key != nil {
+	if cmsg.key != nil && h.msgFields.Key {
 		msg.Key = C.GoBytes(unsafe.Pointer(cmsg.key), C.int(cmsg.key_len))
 	}
 	msg.TopicPartition.Offset = Offset(cmsg.offset)

kafka/producer.go

@@ -423,6 +423,8 @@ func (p *Producer) Purge(flags int) error {
 //                                     Note: timestamps and headers are not supported with this interface.
 //   go.delivery.reports (bool, true) - Forward per-message delivery reports to the
 //                                      Events() channel.
+//   go.delivery.report.fields (string, all) - Comma separated list of fields to enable for delivery reports.
+//                                             Allowed values: all, none (or empty string), key, value
 //   go.events.channel.size (int, 1000000) - Events().
 //   go.produce.channel.size (int, 1000000) - ProduceChannel() buffer size (in number of messages)
 //   go.logs.channel.enable (bool, false) - Forward log to Logs() channel.
@@ -461,6 +463,17 @@ func NewProducer(conf *ConfigMap) (*Producer, error) {
 	}
 	p.handle.fwdDr = v.(bool)
 
+	v, err = confCopy.extract("go.delivery.report.fields", "all")
+	if err != nil {
+		return nil, err
+	}
+
+	msgFields, err := newMessageFieldsFrom(v)
+	if err != nil {
+		return nil, err
+	}
+	p.handle.msgFields = msgFields
+
 	v, err = confCopy.extract("go.events.channel.size", 1000000)
 	if err != nil {
 		return nil, err

kafka/producer_test.go

@@ -17,6 +17,7 @@
 package kafka
 
 import (
+	"bytes"
 	"context"
 	"encoding/binary"
 	"encoding/json"
@@ -563,3 +564,105 @@ func TestTransactionalAPI(t *testing.T) {
 
 	p.Close()
 }
+
+// TestProducerDeliveryReportFields tests the `go.delivery.report.fields` config setting
+func TestProducerDeliveryReportFields(t *testing.T) {
+	t.Run("none", func(t *testing.T) {
+		runProducerDeliveryReportFieldTest(t, &ConfigMap{
+			"socket.timeout.ms":         10,
+			"message.timeout.ms":        10,
+			"go.delivery.report.fields": "",
+		}, func(expected, actual *Message) {
+			if len(actual.Key) > 0 {
+				t.Errorf("key should not be set")
+			}
+			if len(actual.Value) > 0 {
+				t.Errorf("value should not be set")
+			}
+			if s, ok := actual.Opaque.(*string); ok {
+				if *s != *(expected.Opaque.(*string)) {
+					t.Errorf("Opaque should be \"%v\", not \"%v\"", expected.Opaque, actual.Opaque)
+				}
+			} else {
+				t.Errorf("opaque value should be a string, not \"%v\"", actual.Opaque)
+			}
+		})
+	})
+	t.Run("single", func(t *testing.T) {
+		runProducerDeliveryReportFieldTest(t, &ConfigMap{
+			"socket.timeout.ms":         10,
+			"message.timeout.ms":        10,
+			"go.delivery.report.fields": "key",
+		}, func(expected, actual *Message) {
+			if !bytes.Equal(expected.Key, actual.Key) {
+				t.Errorf("key should be \"%s\", not \"%s\"", expected.Key, actual.Key)
+			}
+			if len(actual.Value) > 0 {
+				t.Errorf("value should not be set")
+			}
+			if s, ok := actual.Opaque.(*string); ok {
+				if *s != *(expected.Opaque.(*string)) {
+					t.Errorf("Opaque should be \"%v\", not \"%v\"", expected.Opaque, actual.Opaque)
+				}
+			} else {
+				t.Errorf("opaque value should be a string, not \"%v\"", actual.Opaque)
+			}
+		})
+	})
+	t.Run("multiple", func(t *testing.T) {
+		runProducerDeliveryReportFieldTest(t, &ConfigMap{
+			"socket.timeout.ms":         10,
+			"message.timeout.ms":        10,
+			"go.delivery.report.fields": "key,value",
+		}, func(expected, actual *Message) {
+			if !bytes.Equal(expected.Key, actual.Key) {
+				t.Errorf("key should be \"%s\", not \"%s\"", expected.Key, actual.Key)
+			}
+			if !bytes.Equal(expected.Value, actual.Value) {
+				t.Errorf("value should be \"%s\", not \"%s\"", expected.Value, actual.Value)
+			}
+			if s, ok := actual.Opaque.(*string); ok {
+				if *s != *(expected.Opaque.(*string)) {
+					t.Errorf("Opaque should be \"%v\", not \"%v\"", expected.Opaque, actual.Opaque)
+				}
+			} else {
+				t.Errorf("opaque value should be a string, not \"%v\"", actual.Opaque)
+			}
+		})
+	})
+}
+
+func runProducerDeliveryReportFieldTest(t *testing.T, config *ConfigMap, fn func(expected, actual *Message)) {
+	p, err := NewProducer(config)
+	if err != nil {
+		t.Fatalf("%s", err)
+	}
+
+	topic1 := "gotest"
+
+	myOpq := "My opaque"
+	expected := &Message{
+		TopicPartition: TopicPartition{Topic: &topic1, Partition: 0},
+		Opaque:         &myOpq,
+		Value:          []byte("ProducerChannel"),
+		Key:            []byte("This is my key"),
+	}
+	p.ProduceChannel() <- expected
+
+	// We should expect a single message and possibly events
+	for msgCnt := 0; msgCnt < 1; {
+		ev := <-p.Events()
+		switch e := ev.(type) {
+		case *Message:
+			msgCnt++
+			fn(expected, e)
+		default:
+			t.Logf("Ignored event %s", e)
+		}
+	}
+
+	r := p.Flush(2000)
+	if r > 0 {
+		t.Errorf("Expected empty queue after Flush, still has %d", r)
+	}
+}