otelcol/exporter/kafka: key jaeger messages on traceid

exporter/kafkaexporter/README.md

@@ -13,8 +13,8 @@ The following settings can be optionally configured:
 - `encoding` (default = otlp_proto): The encoding of the traces sent to kafka. All available encodings:
   - `otlp_proto`: payload is Protobuf serialized from `ExportTraceServiceRequest` if set as a traces exporter or `ExportMetricsServiceRequest` for metrics.
   - The following encodings are valid *only* for **traces**.
-    - `jaeger_proto`: the payload is serialized to a single Jaeger proto `Span`.
-    - `jaeger_json`: the payload is serialized to a single Jaeger JSON Span using `jsonpb`.
+    - `jaeger_proto`: the payload is serialized to a single Jaeger proto `Span`, and keyed by TraceID.
+    - `jaeger_json`: the payload is serialized to a single Jaeger JSON Span using `jsonpb`, and keyed by TraceID.
 - `auth`
   - `plain_text`
     - `username`: The username to use.

exporter/kafkaexporter/jaeger_marshaller.go

@@ -47,7 +47,8 @@ func (j jaegerMarshaller) Marshal(traces pdata.Traces) ([]Message, error) {
 				errs = append(errs, err)
 				continue
 			}
-			messages = append(messages, Message{Value: bts})
+			key := []byte(span.TraceID.String())
+			messages = append(messages, Message{Value: bts, Key: key})
 		}
 	}
 	return messages, consumererror.CombineErrors(errs)

exporter/kafkaexporter/jaeger_marshaller_test.go

@@ -41,6 +41,7 @@ func TestJaegerMarshaller(t *testing.T) {
 
 	batches[0].Spans[0].Process = batches[0].Process
 	jaegerProtoBytes, err := batches[0].Spans[0].Marshal()
+	messageKey := []byte(batches[0].Spans[0].TraceID.String())
 	require.NoError(t, err)
 	require.NotNil(t, jaegerProtoBytes)
 
@@ -58,7 +59,7 @@ func TestJaegerMarshaller(t *testing.T) {
 				marshaller: jaegerProtoSpanMarshaller{},
 			},
 			encoding: "jaeger_proto",
-			messages: []Message{{Value: jaegerProtoBytes}},
+			messages: []Message{{Value: jaegerProtoBytes, Key: messageKey}},
 		},
 		{
 			unmarshaller: jaegerMarshaller{
@@ -67,7 +68,7 @@ func TestJaegerMarshaller(t *testing.T) {
 				},
 			},
 			encoding: "jaeger_json",
-			messages: []Message{{Value: jsonByteBuffer.Bytes()}},
+			messages: []Message{{Value: jsonByteBuffer.Bytes(), Key: messageKey}},
 		},
 	}
 	for _, test := range tests {

exporter/kafkaexporter/kafka_exporter.go

@@ -148,6 +148,7 @@ func producerMessages(messages []Message, topic string) []*sarama.ProducerMessag
 		producerMessages[i] = &sarama.ProducerMessage{
 			Topic: topic,
 			Value: sarama.ByteEncoder(messages[i].Value),
+			Key:   sarama.ByteEncoder(messages[i].Key),
 		}
 	}
 	return producerMessages

exporter/kafkaexporter/marshaller.go

@@ -39,6 +39,7 @@ type MetricsMarshaller interface {
 // Message encapsulates Kafka's message payload.
 type Message struct {
 	Value []byte
+	Key   []byte
 }
 
 // tracesMarshallers returns map of supported encodings with TracesMarshaller.