[libbeat] kafka message support headers feature

CHANGELOG.next.asciidoc

@@ -91,6 +91,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...main[Check the HEAD dif
 - Add `script` processor to all beats {issue}29269[29269] {pull}29752[29752]
 - Only connect to Elasticsearch instances with the same version or newer. {pull}29683[29683]
 - Move umask from code to service files. {pull}29708[29708]
+- Add support for kafka message headers. {pull}29940[29940]
 - Add FIPS configuration option for all AWS API calls. {pull}[28899]
 - Add metadata change support for some processors {pull}30183[30183]
 

libbeat/outputs/kafka/client.go

@@ -53,6 +53,8 @@ type client struct {
 
 	producer sarama.AsyncProducer
 
+	recordHeaders []sarama.RecordHeader
+
 	wg sync.WaitGroup
 }
 
@@ -76,7 +78,8 @@ func newKafkaClient(
 	index string,
 	key *fmtstr.EventFormatString,
 	topic outil.Selector,
-	writer codec.Codec,
+	headers map[string]string,
+    writer codec.Codec,
 	cfg *sarama.Config,
 ) (*client, error) {
 	c := &client{
@@ -90,6 +93,20 @@ func newKafkaClient(
 		config:   *cfg,
 		done:     make(chan struct{}),
 	}
+
+	if len(headers) != 0 {
+		recordHeaders := make([]sarama.RecordHeader, 0)
+		for k, v := range headers {
+			recordHeader := sarama.RecordHeader{
+				Key:   []byte(k),
+				Value: []byte(v),
+			}
+
+			recordHeaders = append(recordHeaders, recordHeader)
+		}
+		c.recordHeaders = recordHeaders
+	}
+
 	return c, nil
 }
 

libbeat/outputs/kafka/config.go

@@ -62,6 +62,7 @@ type kafkaConfig struct {
 	BulkMaxSize        int                       `config:"bulk_max_size"`
 	BulkFlushFrequency time.Duration             `config:"bulk_flush_frequency"`
 	MaxRetries         int                       `config:"max_retries"         validate:"min=-1,nonzero"`
+	Headers            map[string]string         `config:"headers"`
 	Backoff            backoffConfig             `config:"backoff"`
 	ClientID           string                    `config:"client_id"`
 	ChanBufferSize     int                       `config:"channel_buffer_size" validate:"min=1"`
@@ -125,6 +126,7 @@ func defaultConfig() kafkaConfig {
 		CompressionLevel: 4,
 		Version:          kafka.Version("1.0.0"),
 		MaxRetries:       3,
+		Headers:          nil,
 		Backoff: backoffConfig{
 			Init: 1 * time.Second,
 			Max:  60 * time.Second,

libbeat/outputs/kafka/kafka.go

@@ -72,7 +72,7 @@ func makeKafka(
 		return outputs.Fail(err)
 	}
 
-	client, err := newKafkaClient(observer, hosts, beat.IndexPrefix, config.Key, topic, codec, libCfg)
+	client, err := newKafkaClient(observer, hosts, beat.IndexPrefix, config.Key, topic, config.Headers, codec, libCfg)
 	if err != nil {
 		return outputs.Fail(err)
 	}

libbeat/outputs/kafka/message.go

@@ -49,5 +49,6 @@ func (m *message) initProducerMessage() {
 		Key:       sarama.ByteEncoder(m.key),
 		Value:     sarama.ByteEncoder(m.value),
 		Timestamp: m.ts,
+		Headers :  m.ref.client.recordHeaders,
 	}
 }