impl log partition by trace id (#43029)

### Description
Added log partitioning capability to the Kafka Exporter. When the
`partition_logs_by_trace_id` option is enabled, log records are
distributed to Kafka partitions based on their TraceID. Records without
a TraceID fall back to the default partitioning strategy. If
`partition_logs_by_resource_attributes` is also enabled, resource
attribute partitioning takes precedence over TraceID partitioning.

### Link to tracking issue
Fixes #39146

### Testing
- Added unit tests to validate TraceID-based log partitioning and option
precedence logic.
- Verified correct partitioning for various combinations of TraceIDs and
resource attributes.

### Documentation
- Updated `exporter/kafkaexporter/README.md` with usage details and
examples for `partition_logs_by_trace_id`.
- Included related changelog updates.

---------

Co-authored-by: Andrew Wilkins <[email protected]>

Author: OhJuhun

.chloggen/impl-39146-log-partition-by-trace-id.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: exporter/kafka
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add support for partitioning log records by trace ID
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [39146]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

exporter/kafkaexporter/README.md

@@ -54,6 +54,7 @@ The following settings can be optionally configured:
 - `partition_traces_by_id` (default = false): configures the exporter to include the trace ID as the message key in trace messages sent to kafka. *Please note:* this setting does not have any effect on Jaeger encoding exporters since Jaeger exporters include trace ID as the message key by default.
 - `partition_metrics_by_resource_attributes` (default = false)  configures the exporter to include the hash of sorted resource attributes as the message partitioning key in metric messages sent to kafka.
 - `partition_logs_by_resource_attributes` (default = false)  configures the exporter to include the hash of sorted resource attributes as the message partitioning key in log messages sent to kafka.
+- - `partition_logs_by_trace_id` (default = false): configures the exporter to partition log messages by trace ID, if the log record has one associated. Note: `partition_logs_by_resource_attributes` and `partition_logs_by_trace_id` are mutually exclusive, and enabling both will lead to an error.
 - `tls`: see [TLS Configuration Settings](https://github.com/open-telemetry/opentelemetry-collector/blob/main/config/configtls/README.md) for the full set of available options.
 - `auth`
   - `plain_text` (Deprecated in v0.123.0: use sasl with mechanism set to PLAIN instead.)

exporter/kafkaexporter/config.go

@@ -4,6 +4,8 @@
 package kafkaexporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/kafkaexporter"
 
 import (
+	"errors"
+
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/configretry"
 	"go.opentelemetry.io/collector/confmap"
@@ -14,6 +16,8 @@ import (
 
 var _ component.Config = (*Config)(nil)
 
+var errLogsPartitionExclusive = errors.New("partition_logs_by_resource_attributes and partition_logs_by_trace_id cannot both be enabled")
+
 // Config defines configuration for Kafka exporter.
 type Config struct {
 	TimeoutSettings           exporterhelper.TimeoutConfig    `mapstructure:",squash"` // squash ensures fields are correctly decoded in embedded struct.
@@ -71,6 +75,21 @@ type Config struct {
 	// If this is true, then the message key will be set to a hash of the resource's identifying
 	// attributes.
 	PartitionLogsByResourceAttributes bool `mapstructure:"partition_logs_by_resource_attributes"`
+
+	// PartitionLogsByTraceID controls partitioning of log messages by trace ID only.
+	// When enabled, the exporter splits incoming logs per TraceID (using SplitLogs)
+	// and sets the Kafka message key to the 16-byte hex string of that TraceID.
+	// If a LogRecord has an empty TraceID, the key may be empty and partition
+	// selection falls back to the Kafka client’s default strategy. Resource
+	// attributes are not used for the key when this option is enabled.
+	PartitionLogsByTraceID bool `mapstructure:"partition_logs_by_trace_id"`
+}
+
+func (c *Config) Validate() (err error) {
+	if c.PartitionLogsByResourceAttributes && c.PartitionLogsByTraceID {
+		return errLogsPartitionExclusive
+	}
+	return err
 }
 
 func (c *Config) Unmarshal(conf *confmap.Conf) error {

exporter/kafkaexporter/config_test.go

@@ -80,6 +80,7 @@ func TestLoadConfig(t *testing.T) {
 				PartitionTracesByID:                  true,
 				PartitionMetricsByResourceAttributes: true,
 				PartitionLogsByResourceAttributes:    true,
+				PartitionLogsByTraceID:               false,
 			},
 		},
 		{
@@ -152,3 +153,34 @@ func TestLoadConfig(t *testing.T) {
 		})
 	}
 }
+
+func TestLoadConfigFailed(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		id            component.ID
+		expectedError error
+		configFile    string
+	}{
+		{
+			id:            component.NewIDWithName(metadata.Type, ""),
+			expectedError: errLogsPartitionExclusive,
+			configFile:    "config-partitioning-failed.yaml",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.id.String(), func(t *testing.T) {
+			cm, err := confmaptest.LoadConf(filepath.Join("testdata", tt.configFile))
+			require.NoError(t, err)
+
+			cfg := createDefaultConfig().(*Config)
+
+			sub, err := cm.Sub(tt.id.String())
+			require.NoError(t, err)
+			require.NoError(t, sub.Unmarshal(cfg))
+
+			assert.ErrorIs(t, xconfmap.Validate(cfg), tt.expectedError)
+		})
+	}
+}

exporter/kafkaexporter/factory.go

@@ -32,6 +32,8 @@ const (
 	defaultPartitionMetricsByResourceAttributesEnabled = false
 	// partitioning logs by resource attributes is disabled by default
 	defaultPartitionLogsByResourceAttributesEnabled = false
+	// partitioning logs by trace id is disabled by default
+	defaultPartitionLogsByTraceIDEnabled = false
 )
 
 // NewFactory creates Kafka exporter factory.
@@ -71,6 +73,7 @@ func createDefaultConfig() component.Config {
 		},
 		PartitionMetricsByResourceAttributes: defaultPartitionMetricsByResourceAttributesEnabled,
 		PartitionLogsByResourceAttributes:    defaultPartitionLogsByResourceAttributesEnabled,
+		PartitionLogsByTraceID:               defaultPartitionLogsByTraceIDEnabled,
 	}
 }
 

exporter/kafkaexporter/kafka_exporter.go

@@ -220,7 +220,7 @@ func (e *kafkaTracesMessenger) marshalData(td ptrace.Traces) ([]marshaler.Messag
 }
 
 func (e *kafkaTracesMessenger) getTopic(ctx context.Context, td ptrace.Traces) string {
-	return getTopic(ctx, e.config.Traces, e.config.TopicFromAttribute, td.ResourceSpans())
+	return getTopic[ptrace.ResourceSpans](ctx, e.config.Traces, e.config.TopicFromAttribute, td.ResourceSpans())
 }
 
 func (e *kafkaTracesMessenger) partitionData(td ptrace.Traces) iter.Seq2[[]byte, ptrace.Traces] {
@@ -265,23 +265,36 @@ func (e *kafkaLogsMessenger) marshalData(ld plog.Logs) ([]marshaler.Message, err
 }
 
 func (e *kafkaLogsMessenger) getTopic(ctx context.Context, ld plog.Logs) string {
-	return getTopic(ctx, e.config.Logs, e.config.TopicFromAttribute, ld.ResourceLogs())
+	return getTopic[plog.ResourceLogs](ctx, e.config.Logs, e.config.TopicFromAttribute, ld.ResourceLogs())
 }
 
 func (e *kafkaLogsMessenger) partitionData(ld plog.Logs) iter.Seq2[[]byte, plog.Logs] {
 	return func(yield func([]byte, plog.Logs) bool) {
-		if !e.config.PartitionLogsByResourceAttributes {
-			yield(nil, ld)
+		if e.config.PartitionLogsByResourceAttributes {
+			for _, resourceLogs := range ld.ResourceLogs().All() {
+				hash := pdatautil.MapHash(resourceLogs.Resource().Attributes())
+				newLogs := plog.NewLogs()
+				resourceLogs.CopyTo(newLogs.ResourceLogs().AppendEmpty())
+				if !yield(hash[:], newLogs) {
+					return
+				}
+			}
 			return
 		}
-		for _, resourceLogs := range ld.ResourceLogs().All() {
-			hash := pdatautil.MapHash(resourceLogs.Resource().Attributes())
-			newLogs := plog.NewLogs()
-			resourceLogs.CopyTo(newLogs.ResourceLogs().AppendEmpty())
-			if !yield(hash[:], newLogs) {
-				return
+		if e.config.PartitionLogsByTraceID {
+			for _, l := range batchpersignal.SplitLogs(ld) {
+				var key []byte
+				traceID := l.ResourceLogs().At(0).ScopeLogs().At(0).LogRecords().At(0).TraceID()
+				if !traceID.IsEmpty() {
+					key = []byte(traceutil.TraceIDToHexOrEmptyString(traceID))
+				}
+				if !yield(key, l) {
+					return
+				}
 			}
+			return
 		}
+		yield(nil, ld)
 	}
 }
 
@@ -308,7 +321,7 @@ func (e *kafkaMetricsMessenger) marshalData(md pmetric.Metrics) ([]marshaler.Mes
 }
 
 func (e *kafkaMetricsMessenger) getTopic(ctx context.Context, md pmetric.Metrics) string {
-	return getTopic(ctx, e.config.Metrics, e.config.TopicFromAttribute, md.ResourceMetrics())
+	return getTopic[pmetric.ResourceMetrics](ctx, e.config.Metrics, e.config.TopicFromAttribute, md.ResourceMetrics())
 }
 
 func (e *kafkaMetricsMessenger) partitionData(md pmetric.Metrics) iter.Seq2[[]byte, pmetric.Metrics] {
@@ -351,7 +364,7 @@ func (e *kafkaProfilesMessenger) marshalData(ld pprofile.Profiles) ([]marshaler.
 }
 
 func (e *kafkaProfilesMessenger) getTopic(ctx context.Context, ld pprofile.Profiles) string {
-	return getTopic(ctx, e.config.Profiles, e.config.TopicFromAttribute, ld.ResourceProfiles())
+	return getTopic[pprofile.ResourceProfiles](ctx, e.config.Profiles, e.config.TopicFromAttribute, ld.ResourceProfiles())
 }
 
 func (*kafkaProfilesMessenger) partitionData(ld pprofile.Profiles) iter.Seq2[[]byte, pprofile.Profiles] {

exporter/kafkaexporter/kafka_exporter_test.go

@@ -30,6 +30,7 @@ import (
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/kafkaexporter/internal/kafkaclient"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/kafkaexporter/internal/metadata"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/traceutil"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/kafka/kafkatest"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/kafka/topic"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest/plogtest"
@@ -982,6 +983,90 @@ func TestLogsPusher_partitioning(t *testing.T) {
 		assert.Equal(t, keys[0], keys[1])
 		assert.NotEqual(t, keys[0], keys[2])
 	})
+	t.Run("trace_id_partitioning", func(t *testing.T) {
+		config := createDefaultConfig().(*Config)
+		config.PartitionLogsByTraceID = true
+		exp, producer := newMockLogsExporter(t, *config, componenttest.NewNopHost())
+
+		// Build input with three ResourceLogs: two share the same TraceID, one has a different TraceID.
+		in := plog.NewLogs()
+		var rls []plog.ResourceLogs
+		tid1 := pcommon.TraceID([16]byte{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1})
+		tid2 := pcommon.TraceID([16]byte{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 2})
+
+		makeResourceLogs := func(tid pcommon.TraceID) plog.ResourceLogs {
+			rl := in.ResourceLogs().AppendEmpty()
+			rl.Resource().Attributes().PutStr("service.name", "svc")
+			sl := rl.ScopeLogs().AppendEmpty()
+			lr := sl.LogRecords().AppendEmpty()
+			lr.SetTraceID(tid)
+			return rl
+		}
+		rl1 := makeResourceLogs(tid1)
+		rl2 := makeResourceLogs(tid1)
+		rl3 := makeResourceLogs(tid2)
+		rls = append(rls, rl1, rl2, rl3)
+
+		var keys [][]byte
+		for i := 0; i < len(rls); i++ {
+			producer.ExpectSendMessageWithMessageCheckerFunctionAndSucceed(
+				func(msg *sarama.ProducerMessage) error {
+					value, err := msg.Value.Encode()
+					require.NoError(t, err)
+					output, err := (&plog.ProtoUnmarshaler{}).UnmarshalLogs(value)
+					require.NoError(t, err)
+
+					require.Equal(t, 1, output.ResourceLogs().Len())
+					assert.NoError(t, plogtest.CompareResourceLogs(
+						rls[i],
+						output.ResourceLogs().At(0),
+					))
+
+					key, err := msg.Key.Encode()
+					require.NoError(t, err)
+					keys = append(keys, key)
+					return nil
+				},
+			)
+		}
+
+		err := exp.exportData(t.Context(), in)
+		require.NoError(t, err)
+
+		require.Len(t, keys, 3)
+		// Keys should be the hex TraceID only, identical for same TraceID, different otherwise.
+		expected1 := []byte(traceutil.TraceIDToHexOrEmptyString(tid1))
+		expected2 := []byte(traceutil.TraceIDToHexOrEmptyString(tid2))
+		assert.Equal(t, expected1, keys[0])
+		assert.Equal(t, expected1, keys[1])
+		assert.Equal(t, expected2, keys[2])
+		assert.NotEqual(t, keys[0], keys[2])
+	})
+
+	// ensure that when TraceID partitioning is enabled but a log record has no TraceID,
+	// the exporter falls back to default partitioning (nil key).
+	t.Run("trace_id_partitioning_missing_traceid_defaults_to_nil_key", func(t *testing.T) {
+		config := createDefaultConfig().(*Config)
+		config.PartitionLogsByTraceID = true
+		exp, producer := newMockLogsExporter(t, *config, componenttest.NewNopHost())
+
+		in := plog.NewLogs()
+		rl := in.ResourceLogs().AppendEmpty()
+		rl.Resource().Attributes().PutStr("service.name", "svc")
+		sl := rl.ScopeLogs().AppendEmpty()
+		_ = sl.LogRecords().AppendEmpty()
+
+		producer.ExpectSendMessageWithMessageCheckerFunctionAndSucceed(
+			func(msg *sarama.ProducerMessage) error {
+				if msg.Key != nil {
+					return fmt.Errorf("expected nil key for missing TraceID, got %v", msg.Key)
+				}
+				return nil
+			},
+		)
+
+		require.NoError(t, exp.exportData(t.Context(), in))
+	})
 }
 
 func TestProfilesPusher(t *testing.T) {
@@ -1379,11 +1464,11 @@ func Test_GetTopic(t *testing.T) {
 			topic := ""
 			switch r := tests[i].resource.(type) {
 			case pmetric.ResourceMetricsSlice:
-				topic = getTopic(tests[i].ctx, tests[i].signalCfg, tests[i].topicFromAttribute, r)
+				topic = getTopic[pmetric.ResourceMetrics](tests[i].ctx, tests[i].signalCfg, tests[i].topicFromAttribute, r)
 			case ptrace.ResourceSpansSlice:
-				topic = getTopic(tests[i].ctx, tests[i].signalCfg, tests[i].topicFromAttribute, r)
+				topic = getTopic[ptrace.ResourceSpans](tests[i].ctx, tests[i].signalCfg, tests[i].topicFromAttribute, r)
 			case plog.ResourceLogsSlice:
-				topic = getTopic(tests[i].ctx, tests[i].signalCfg, tests[i].topicFromAttribute, r)
+				topic = getTopic[plog.ResourceLogs](tests[i].ctx, tests[i].signalCfg, tests[i].topicFromAttribute, r)
 			}
 			assert.Equal(t, tests[i].wantTopic, topic)
 		})

exporter/kafkaexporter/testdata/config-partitioning-failed.yaml

@@ -0,0 +1,32 @@
+kafka:
+  topic: spans
+  brokers:
+    - "foo:123"
+    - "bar:456"
+  producer:
+    max_message_bytes: 10000000
+    required_acks: -1 # WaitForAll
+  timeout: 10s
+  partition_traces_by_id: true
+  partition_metrics_by_resource_attributes: true
+  partition_logs_by_resource_attributes: true
+  partition_logs_by_trace_id: true
+  sending_queue:
+    enabled: true
+    num_consumers: 2
+    queue_size: 10
+  retry_on_failure:
+    enabled: true
+    initial_interval: 10s
+    max_interval: 60s
+    max_elapsed_time: 10m
+kafka/legacy_topic:
+  topic: legacy_topic
+  metrics:
+    topic: metrics_topic
+  logs:
+    topic_from_metadata_key: metadata_key
+kafka/legacy_encoding:
+  encoding: legacy_encoding
+  metrics:
+    encoding: metrics_encoding