Feature/kakfa exporter - kafka key by TraceID

exporter/kafkaexporter/go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/gogo/protobuf v1.3.2
 	github.com/jaegertracing/jaeger v1.41.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal v0.83.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/batchpersignal v0.83.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/jaeger v0.83.0
 	github.com/stretchr/testify v1.8.4
 	github.com/xdg-go/scram v1.1.2
@@ -82,6 +83,8 @@ require (
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal => ../../internal/coreinternal
 
+replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/batchpersignal => ../../pkg/batchpersignal
+
 replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/jaeger => ../../pkg/translator/jaeger
 
 retract (

exporter/kafkaexporter/pdata_marshaler.go

@@ -8,6 +8,8 @@ import (
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.opentelemetry.io/collector/pdata/ptrace"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/batchpersignal"
 )
 
 type pdataLogsMarshaler struct {
@@ -74,16 +76,22 @@ type pdataTracesMarshaler struct {
 }
 
 func (p pdataTracesMarshaler) Marshal(td ptrace.Traces, topic string) ([]*sarama.ProducerMessage, error) {
-	bts, err := p.marshaler.MarshalTraces(td)
-	if err != nil {
-		return nil, err
-	}
-	return []*sarama.ProducerMessage{
-		{
+	var messages []*sarama.ProducerMessage
+
+	for _, tracesById := range batchpersignal.SplitTraces(td) {
+		bts, err := p.marshaler.MarshalTraces(tracesById)
+		if err != nil {
+			return nil, err
+		}
+		var traceId = tracesById.ResourceSpans().At(0).ScopeSpans().At(0).Spans().At(0).TraceID()
+		key := []byte(traceId.String())
+		messages = append(messages, &sarama.ProducerMessage{
 			Topic: topic,
 			Value: sarama.ByteEncoder(bts),
-		},
-	}, nil
+			Key:   sarama.ByteEncoder(key),
+		})
+	}
+	return messages, nil
 }
 
 func (p pdataTracesMarshaler) Encoding() string {