From 943c736d161051a1a1c0d527d6aaf1b696c1d8ab Mon Sep 17 00:00:00 2001
From: Bogdan Drutu <bogdandrutu@gmail.com>
Date: Mon, 16 Sep 2024 09:29:31 -0700
Subject: [PATCH 01/10] [transformprocessor]: Remove unnecessary data copy when
 transform sum to/from gauge (#35177)

Signed-off-by: Bogdan Drutu <bogdandrutu@gmail.com>
---
 .chloggen/rm-copy-transformprocessor.yaml     | 22 +++++++++++++++++++
 .../metrics/func_convert_gauge_to_sum.go      |  4 ++--
 .../func_convert_gauge_to_sum_datapoint.go    |  4 ++--
 .../metrics/func_convert_sum_to_gauge.go      |  2 +-
 .../func_convert_sum_to_gauge_datapoint.go    |  4 ++--
 .../metrics/func_extract_sum_metric.go        |  2 +-
 6 files changed, 30 insertions(+), 8 deletions(-)
 create mode 100644 .chloggen/rm-copy-transformprocessor.yaml

diff --git a/.chloggen/rm-copy-transformprocessor.yaml b/.chloggen/rm-copy-transformprocessor.yaml
new file mode 100644
index 000000000000..074c499d91c7
--- /dev/null
+++ b/.chloggen/rm-copy-transformprocessor.yaml
@@ -0,0 +1,22 @@
+# 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: transformprocessor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Remove unnecessary data copy when transform sum to/from gauge
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [35177]
+
+# 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]
diff --git a/processor/transformprocessor/internal/metrics/func_convert_gauge_to_sum.go b/processor/transformprocessor/internal/metrics/func_convert_gauge_to_sum.go
index 3c3a5100dc3c..3fc352f0faca 100644
--- a/processor/transformprocessor/internal/metrics/func_convert_gauge_to_sum.go
+++ b/processor/transformprocessor/internal/metrics/func_convert_gauge_to_sum.go
@@ -54,8 +54,8 @@ func convertGaugeToSum(stringAggTemp string, monotonic bool) (ottl.ExprFunc[ottl
 		metric.SetEmptySum().SetAggregationTemporality(aggTemp)
 		metric.Sum().SetIsMonotonic(monotonic)
 
-		// Setting the data type removed all the data points, so we must copy them back to the metric.
-		dps.CopyTo(metric.Sum().DataPoints())
+		// Setting the data type removed all the data points, so we must move them back to the metric.
+		dps.MoveAndAppendTo(metric.Sum().DataPoints())
 
 		return nil, nil
 	}, nil
diff --git a/processor/transformprocessor/internal/metrics/func_convert_gauge_to_sum_datapoint.go b/processor/transformprocessor/internal/metrics/func_convert_gauge_to_sum_datapoint.go
index dab12e96d094..61c848ab1b52 100644
--- a/processor/transformprocessor/internal/metrics/func_convert_gauge_to_sum_datapoint.go
+++ b/processor/transformprocessor/internal/metrics/func_convert_gauge_to_sum_datapoint.go
@@ -50,8 +50,8 @@ func convertDatapointGaugeToSum(stringAggTemp string, monotonic bool) (ottl.Expr
 		metric.SetEmptySum().SetAggregationTemporality(aggTemp)
 		metric.Sum().SetIsMonotonic(monotonic)
 
-		// Setting the data type removed all the data points, so we must copy them back to the metric.
-		dps.CopyTo(metric.Sum().DataPoints())
+		// Setting the data type removed all the data points, so we must move them back to the metric.
+		dps.MoveAndAppendTo(metric.Sum().DataPoints())
 
 		return nil, nil
 	}, nil
diff --git a/processor/transformprocessor/internal/metrics/func_convert_sum_to_gauge.go b/processor/transformprocessor/internal/metrics/func_convert_sum_to_gauge.go
index f4763e65c9e5..212395bd524b 100644
--- a/processor/transformprocessor/internal/metrics/func_convert_sum_to_gauge.go
+++ b/processor/transformprocessor/internal/metrics/func_convert_sum_to_gauge.go
@@ -30,7 +30,7 @@ func convertSumToGauge() (ottl.ExprFunc[ottlmetric.TransformContext], error) {
 		dps := metric.Sum().DataPoints()
 
 		// Setting the data type removed all the data points, so we must copy them back to the metric.
-		dps.CopyTo(metric.SetEmptyGauge().DataPoints())
+		dps.MoveAndAppendTo(metric.SetEmptyGauge().DataPoints())
 
 		return nil, nil
 	}, nil
diff --git a/processor/transformprocessor/internal/metrics/func_convert_sum_to_gauge_datapoint.go b/processor/transformprocessor/internal/metrics/func_convert_sum_to_gauge_datapoint.go
index ca2f09c8a121..1943d2d9796a 100644
--- a/processor/transformprocessor/internal/metrics/func_convert_sum_to_gauge_datapoint.go
+++ b/processor/transformprocessor/internal/metrics/func_convert_sum_to_gauge_datapoint.go
@@ -29,8 +29,8 @@ func convertDatapointSumToGauge() (ottl.ExprFunc[ottldatapoint.TransformContext]
 
 		dps := metric.Sum().DataPoints()
 
-		// Setting the data type removed all the data points, so we must copy them back to the metric.
-		dps.CopyTo(metric.SetEmptyGauge().DataPoints())
+		// Setting the data type removed all the data points, so we must move them back to the metric.
+		dps.MoveAndAppendTo(metric.SetEmptyGauge().DataPoints())
 
 		return nil, nil
 	}, nil
diff --git a/processor/transformprocessor/internal/metrics/func_extract_sum_metric.go b/processor/transformprocessor/internal/metrics/func_extract_sum_metric.go
index 78b47623cdf3..f002260944ac 100644
--- a/processor/transformprocessor/internal/metrics/func_extract_sum_metric.go
+++ b/processor/transformprocessor/internal/metrics/func_extract_sum_metric.go
@@ -32,7 +32,7 @@ func createExtractSumMetricFunction(_ ottl.FunctionContext, oArgs ottl.Arguments
 	return extractSumMetric(args.Monotonic)
 }
 
-// this interface helps unify the logic for extracting data from different histogram types
+// SumCountDataPoint interface helps unify the logic for extracting data from different histogram types
 // all supported metric types' datapoints implement it
 type SumCountDataPoint interface {
 	Attributes() pcommon.Map

From af5b729e12825cd73c1c1812bc6a85570029d502 Mon Sep 17 00:00:00 2001
From: Bogdan Drutu <bogdandrutu@gmail.com>
Date: Mon, 16 Sep 2024 10:34:24 -0700
Subject: [PATCH 02/10] Remove unnecessary copies in deltatorateprocessor
 (#35165)

Signed-off-by: Bogdan Drutu <bogdandrutu@gmail.com>
---
 .chloggen/rm-copies-deltatorateprocessor.yaml | 22 ++++++++++++++
 processor/deltatorateprocessor/processor.go   | 29 +++++++------------
 2 files changed, 33 insertions(+), 18 deletions(-)
 create mode 100644 .chloggen/rm-copies-deltatorateprocessor.yaml

diff --git a/.chloggen/rm-copies-deltatorateprocessor.yaml b/.chloggen/rm-copies-deltatorateprocessor.yaml
new file mode 100644
index 000000000000..5e26c4fbcb51
--- /dev/null
+++ b/.chloggen/rm-copies-deltatorateprocessor.yaml
@@ -0,0 +1,22 @@
+# 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: deltatorateprocessor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Remove unnecessary data copies.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [35165]
+
+# 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]
diff --git a/processor/deltatorateprocessor/processor.go b/processor/deltatorateprocessor/processor.go
index 8da2382ef0ee..a9ba3aca4697 100644
--- a/processor/deltatorateprocessor/processor.go
+++ b/processor/deltatorateprocessor/processor.go
@@ -55,33 +55,26 @@ func (dtrp *deltaToRateProcessor) processMetrics(_ context.Context, md pmetric.M
 					dtrp.logger.Info(fmt.Sprintf("Configured metric for rate calculation %s is not a delta sum\n", metric.Name()))
 					continue
 				}
-				newDoubleDataPointSlice := pmetric.NewNumberDataPointSlice()
-				dataPoints := metric.Sum().DataPoints()
+				dataPointSlice := metric.Sum().DataPoints()
 
-				for i := 0; i < dataPoints.Len(); i++ {
-					fromDataPoint := dataPoints.At(i)
-					newDp := newDoubleDataPointSlice.AppendEmpty()
-					fromDataPoint.CopyTo(newDp)
+				for i := 0; i < dataPointSlice.Len(); i++ {
+					dataPoint := dataPointSlice.At(i)
 
-					durationNanos := time.Duration(fromDataPoint.Timestamp() - fromDataPoint.StartTimestamp())
+					durationNanos := time.Duration(dataPoint.Timestamp() - dataPoint.StartTimestamp())
 					var rate float64
-					switch fromDataPoint.ValueType() {
+					switch dataPoint.ValueType() {
 					case pmetric.NumberDataPointValueTypeDouble:
-						rate = calculateRate(fromDataPoint.DoubleValue(), durationNanos)
+						rate = calculateRate(dataPoint.DoubleValue(), durationNanos)
 					case pmetric.NumberDataPointValueTypeInt:
-						rate = calculateRate(float64(fromDataPoint.IntValue()), durationNanos)
+						rate = calculateRate(float64(dataPoint.IntValue()), durationNanos)
 					default:
-						return md, consumererror.NewPermanent(fmt.Errorf("invalid data point type:%d", fromDataPoint.ValueType()))
+						return md, consumererror.NewPermanent(fmt.Errorf("invalid data point type:%d", dataPoint.ValueType()))
 					}
-					newDp.SetDoubleValue(rate)
+					dataPoint.SetDoubleValue(rate)
 				}
 
-				dps := metric.SetEmptyGauge().DataPoints()
-				dps.EnsureCapacity(newDoubleDataPointSlice.Len())
-				for d := 0; d < newDoubleDataPointSlice.Len(); d++ {
-					dp := dps.AppendEmpty()
-					newDoubleDataPointSlice.At(d).CopyTo(dp)
-				}
+				// Setting the data type removed all the data points, so we must move them back to the metric.
+				dataPointSlice.MoveAndAppendTo(metric.SetEmptyGauge().DataPoints())
 			}
 		}
 	}

From 66f6f0670bf95dcae676add1e59f7fc183a960bd Mon Sep 17 00:00:00 2001
From: Shaun Remekie <shaun.remekie@coralogix.com>
Date: Mon, 16 Sep 2024 20:44:57 +0200
Subject: [PATCH 03/10] [chore] [exporter/coralogix] update domain docs
 (#35200)

**Description:** <Describe what has changed.>
<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->

update docss with newly added domain.

**Link to tracking Issue:** <Issue number if applicable>

**Testing:** <Describe what testing was performed and which tests were
added.>

**Documentation:** <Describe the documentation added.>
---
 exporter/coralogixexporter/README.md | 17 +++++++++--------
 1 file changed, 9 insertions(+), 8 deletions(-)

diff --git a/exporter/coralogixexporter/README.md b/exporter/coralogixexporter/README.md
index da07aca9cfe0..f6068e6163d9 100644
--- a/exporter/coralogixexporter/README.md
+++ b/exporter/coralogixexporter/README.md
@@ -86,14 +86,15 @@ exporters:
 
 Depending on your region, you might need to use a different domain. Here are the available domains:
 
-| Region  | Domain                        |
-|---------|---------------------------------|
-| USA1    | `coralogix.us`        |
-| USA2    | `cx498.coralogix.com` |
-| APAC1   | `coralogix.in`        |
-| APAC2   | `coralogixsg.com`     |
-| EUROPE1 | `coralogix.com`       |
-| EUROPE2 | `eu2.coralogix.com`   |
+| Region  | Domain                  |
+|---------|-------------------------|
+| USA1    | `coralogix.us`          |
+| USA2    | `cx498.coralogix.com`   |
+| APAC1   | `coralogix.in`          |
+| APAC2   | `coralogixsg.com`       |
+| APAC3   | `ap3.coralogix.com`     |
+| EUROPE1 | `coralogix.com`         |
+| EUROPE2 | `eu2.coralogix.com`     |
 
 Additionally, Coralogix supports AWS PrivateLink, which provides private connectivity between virtual private clouds (VPCs), supported AWS services, and your on-premises networks without exposing your traffic to the public internet.
 

From fa62972b96fa147a53e366d1e6e3487633c9865c Mon Sep 17 00:00:00 2001
From: Tyler Helmuth <12352919+TylerHelmuth@users.noreply.github.com>
Date: Mon, 16 Sep 2024 15:08:24 -0600
Subject: [PATCH 04/10] [chore] Fix main (#35222)

---
 .github/CODEOWNERS                             | 2 +-
 .github/ISSUE_TEMPLATE/bug_report.yaml         | 1 +
 .github/ISSUE_TEMPLATE/feature_request.yaml    | 1 +
 .github/ISSUE_TEMPLATE/other.yaml              | 1 +
 .github/ISSUE_TEMPLATE/unmaintained.yaml       | 1 +
 exporter/dorisexporter/exporter_logs_test.go   | 3 ++-
 exporter/dorisexporter/exporter_traces_test.go | 3 ++-
 exporter/dorisexporter/factory.go              | 2 +-
 pkg/datadog/metadata.yaml                      | 3 +++
 9 files changed, 13 insertions(+), 4 deletions(-)
 create mode 100644 pkg/datadog/metadata.yaml

diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS
index 676015f2efe9..d8b547da9834 100644
--- a/.github/CODEOWNERS
+++ b/.github/CODEOWNERS
@@ -146,8 +146,8 @@ internal/tools/                                                     @open-teleme
 
 pkg/batchperresourceattr/                                           @open-telemetry/collector-contrib-approvers @atoulme @dmitryax
 pkg/batchpersignal/                                                 @open-telemetry/collector-contrib-approvers @jpkrohling
-pkg/experimentalmetricmetadata/                                     @open-telemetry/collector-contrib-approvers @rmfitzpatrick
 pkg/datadog/                                                        @open-telemetry/collector-contrib-approvers @mx-psi @dineshg13 @liustanley @songy23 @mackjmr @ankitpatel96
+pkg/experimentalmetricmetadata/                                     @open-telemetry/collector-contrib-approvers @rmfitzpatrick
 pkg/golden/                                                         @open-telemetry/collector-contrib-approvers @djaglowski @atoulme
 pkg/ottl/                                                           @open-telemetry/collector-contrib-approvers @TylerHelmuth @kentquirk @bogdandrutu @evan-bradley
 pkg/pdatatest/                                                      @open-telemetry/collector-contrib-approvers @djaglowski @fatsheep9146
diff --git a/.github/ISSUE_TEMPLATE/bug_report.yaml b/.github/ISSUE_TEMPLATE/bug_report.yaml
index 5374e66a9a89..9a3355295459 100644
--- a/.github/ISSUE_TEMPLATE/bug_report.yaml
+++ b/.github/ISSUE_TEMPLATE/bug_report.yaml
@@ -142,6 +142,7 @@ body:
       - internal/tools
       - pkg/batchperresourceattr
       - pkg/batchpersignal
+      - pkg/datadog
       - pkg/experimentalmetricmetadata
       - pkg/golden
       - pkg/ottl
diff --git a/.github/ISSUE_TEMPLATE/feature_request.yaml b/.github/ISSUE_TEMPLATE/feature_request.yaml
index cbaad5acb6ee..355b7f41489b 100644
--- a/.github/ISSUE_TEMPLATE/feature_request.yaml
+++ b/.github/ISSUE_TEMPLATE/feature_request.yaml
@@ -136,6 +136,7 @@ body:
       - internal/tools
       - pkg/batchperresourceattr
       - pkg/batchpersignal
+      - pkg/datadog
       - pkg/experimentalmetricmetadata
       - pkg/golden
       - pkg/ottl
diff --git a/.github/ISSUE_TEMPLATE/other.yaml b/.github/ISSUE_TEMPLATE/other.yaml
index 2bb37ab37089..43d556c0b36e 100644
--- a/.github/ISSUE_TEMPLATE/other.yaml
+++ b/.github/ISSUE_TEMPLATE/other.yaml
@@ -136,6 +136,7 @@ body:
       - internal/tools
       - pkg/batchperresourceattr
       - pkg/batchpersignal
+      - pkg/datadog
       - pkg/experimentalmetricmetadata
       - pkg/golden
       - pkg/ottl
diff --git a/.github/ISSUE_TEMPLATE/unmaintained.yaml b/.github/ISSUE_TEMPLATE/unmaintained.yaml
index 55b55fd1b45c..a5ceaea8a0fe 100644
--- a/.github/ISSUE_TEMPLATE/unmaintained.yaml
+++ b/.github/ISSUE_TEMPLATE/unmaintained.yaml
@@ -141,6 +141,7 @@ body:
       - internal/tools
       - pkg/batchperresourceattr
       - pkg/batchpersignal
+      - pkg/datadog
       - pkg/experimentalmetricmetadata
       - pkg/golden
       - pkg/ottl
diff --git a/exporter/dorisexporter/exporter_logs_test.go b/exporter/dorisexporter/exporter_logs_test.go
index 8e28aa1ec73b..60097ab6ce4a 100644
--- a/exporter/dorisexporter/exporter_logs_test.go
+++ b/exporter/dorisexporter/exporter_logs_test.go
@@ -10,6 +10,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/plog"
@@ -52,7 +53,7 @@ func TestPushLogData(t *testing.T) {
 			_, _ = w.Write([]byte(`{"Status":"Success"}`))
 		})
 		err = server.ListenAndServe()
-		require.Equal(t, http.ErrServerClosed, err)
+		assert.Equal(t, http.ErrServerClosed, err)
 	}()
 
 	err0 := fmt.Errorf("Not Started")
diff --git a/exporter/dorisexporter/exporter_traces_test.go b/exporter/dorisexporter/exporter_traces_test.go
index aafc08416132..396c320cf2e8 100644
--- a/exporter/dorisexporter/exporter_traces_test.go
+++ b/exporter/dorisexporter/exporter_traces_test.go
@@ -10,6 +10,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/ptrace"
@@ -52,7 +53,7 @@ func TestPushTraceData(t *testing.T) {
 			_, _ = w.Write([]byte(`{"Status":"Success"}`))
 		})
 		err = server.ListenAndServe()
-		require.Equal(t, http.ErrServerClosed, err)
+		assert.Equal(t, http.ErrServerClosed, err)
 	}()
 
 	err0 := fmt.Errorf("Not Started")
diff --git a/exporter/dorisexporter/factory.go b/exporter/dorisexporter/factory.go
index be7259cb4a9e..2acf0c3f36f2 100644
--- a/exporter/dorisexporter/factory.go
+++ b/exporter/dorisexporter/factory.go
@@ -60,7 +60,7 @@ func createLogsExporter(ctx context.Context, set exporter.Settings, cfg componen
 		exporterhelper.WithStart(exporter.start),
 		exporterhelper.WithShutdown(exporter.shutdown),
 		// we config the timeout option in http client, so we don't need to set timeout here
-		exporterhelper.WithTimeout(exporterhelper.TimeoutSettings{Timeout: 0}),
+		exporterhelper.WithTimeout(exporterhelper.TimeoutConfig{Timeout: 0}),
 		exporterhelper.WithQueue(c.QueueSettings),
 		exporterhelper.WithRetry(c.BackOffConfig),
 	)
diff --git a/pkg/datadog/metadata.yaml b/pkg/datadog/metadata.yaml
new file mode 100644
index 000000000000..dae3d0ab2af6
--- /dev/null
+++ b/pkg/datadog/metadata.yaml
@@ -0,0 +1,3 @@
+status:
+  codeowners:
+    active: [mx-psi,dineshg13, liustanley, songy23, mackjmr, ankitpatel96]
\ No newline at end of file

From e5e9e879cba62ad3b7079f5825cf9021a443835d Mon Sep 17 00:00:00 2001
From: Roger Coll <rogercoll@protonmail.com>
Date: Mon, 16 Sep 2024 23:16:52 +0200
Subject: [PATCH 05/10] chore: remove redundant config Validate call (#35199)

**Description:** <Describe what has changed.>
<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
Configuration validation is done during collector's startup, making it
redundant when being called inside component's logic. This PR removes
the Validate call done during exporter's constructor.

**Link to tracking Issue:** <Issue number if applicable>
https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/33498

**Testing:** <Describe what testing was performed and which tests were
added.> Same coverage after removed lines: `coverage: 77.8% of
statements`

`TestFactory_CreateLogsExporter_Fail` and
`TestFactory_CreateTracesExporter_Fail` functions are covered by the
already available `config_test.go` cases.

**Documentation:** <Describe the documentation added.>
---
 exporter/elasticsearchexporter/config_test.go |  7 +++---
 exporter/elasticsearchexporter/exporter.go    |  8 ++-----
 exporter/elasticsearchexporter/factory.go     | 22 ++++++-------------
 .../elasticsearchexporter/factory_test.go     | 18 ---------------
 4 files changed, 12 insertions(+), 43 deletions(-)

diff --git a/exporter/elasticsearchexporter/config_test.go b/exporter/elasticsearchexporter/config_test.go
index 8bc9de0b5f3f..a27a28ccfe6e 100644
--- a/exporter/elasticsearchexporter/config_test.go
+++ b/exporter/elasticsearchexporter/config_test.go
@@ -395,8 +395,7 @@ func TestConfig_Validate(t *testing.T) {
 
 	for name, tt := range tests {
 		t.Run(name, func(t *testing.T) {
-			err := tt.config.Validate()
-			assert.EqualError(t, err, tt.err)
+			assert.EqualError(t, component.ValidateConfig(tt.config), tt.err)
 		})
 	}
 }
@@ -405,13 +404,13 @@ func TestConfig_Validate_Environment(t *testing.T) {
 	t.Run("valid", func(t *testing.T) {
 		t.Setenv("ELASTICSEARCH_URL", "http://test:9200")
 		config := withDefaultConfig()
-		err := config.Validate()
+		err := component.ValidateConfig(config)
 		require.NoError(t, err)
 	})
 	t.Run("invalid", func(t *testing.T) {
 		t.Setenv("ELASTICSEARCH_URL", "http://valid:9200, *:!")
 		config := withDefaultConfig()
-		err := config.Validate()
+		err := component.ValidateConfig(config)
 		assert.EqualError(t, err, `invalid endpoint "*:!": parse "*:!": first path segment in URL cannot contain colon`)
 	})
 }
diff --git a/exporter/elasticsearchexporter/exporter.go b/exporter/elasticsearchexporter/exporter.go
index 2bf4c0250fa4..6f70a9ac6b84 100644
--- a/exporter/elasticsearchexporter/exporter.go
+++ b/exporter/elasticsearchexporter/exporter.go
@@ -42,11 +42,7 @@ func newExporter(
 	set exporter.Settings,
 	index string,
 	dynamicIndex bool,
-) (*elasticsearchExporter, error) {
-	if err := cfg.Validate(); err != nil {
-		return nil, err
-	}
-
+) *elasticsearchExporter {
 	model := &encodeModel{
 		dedot: cfg.Mapping.Dedot,
 		mode:  cfg.MappingMode(),
@@ -72,7 +68,7 @@ func newExporter(
 		model:          model,
 		logstashFormat: cfg.LogstashFormat,
 		otel:           otel,
-	}, nil
+	}
 }
 
 func (e *elasticsearchExporter) Start(ctx context.Context, host component.Host) error {
diff --git a/exporter/elasticsearchexporter/factory.go b/exporter/elasticsearchexporter/factory.go
index da97ba7d9f05..3f48ca1e2ec7 100644
--- a/exporter/elasticsearchexporter/factory.go
+++ b/exporter/elasticsearchexporter/factory.go
@@ -7,7 +7,6 @@ package elasticsearchexporter // import "github.com/open-telemetry/opentelemetry
 
 import (
 	"context"
-	"fmt"
 	"net/http"
 	"time"
 
@@ -113,10 +112,7 @@ func createLogsExporter(
 	}
 	logConfigDeprecationWarnings(cf, set.Logger)
 
-	exporter, err := newExporter(cf, set, index, cf.LogsDynamicIndex.Enabled)
-	if err != nil {
-		return nil, fmt.Errorf("cannot configure Elasticsearch exporter: %w", err)
-	}
+	exporter := newExporter(cf, set, index, cf.LogsDynamicIndex.Enabled)
 
 	return exporterhelper.NewLogsExporter(
 		ctx,
@@ -135,10 +131,8 @@ func createMetricsExporter(
 	cf := cfg.(*Config)
 	logConfigDeprecationWarnings(cf, set.Logger)
 
-	exporter, err := newExporter(cf, set, cf.MetricsIndex, cf.MetricsDynamicIndex.Enabled)
-	if err != nil {
-		return nil, fmt.Errorf("cannot configure Elasticsearch exporter: %w", err)
-	}
+	exporter := newExporter(cf, set, cf.MetricsIndex, cf.MetricsDynamicIndex.Enabled)
+
 	return exporterhelper.NewMetricsExporter(
 		ctx,
 		set,
@@ -150,15 +144,13 @@ func createMetricsExporter(
 
 func createTracesExporter(ctx context.Context,
 	set exporter.Settings,
-	cfg component.Config) (exporter.Traces, error) {
-
+	cfg component.Config,
+) (exporter.Traces, error) {
 	cf := cfg.(*Config)
 	logConfigDeprecationWarnings(cf, set.Logger)
 
-	exporter, err := newExporter(cf, set, cf.TracesIndex, cf.TracesDynamicIndex.Enabled)
-	if err != nil {
-		return nil, fmt.Errorf("cannot configure Elasticsearch exporter: %w", err)
-	}
+	exporter := newExporter(cf, set, cf.TracesIndex, cf.TracesDynamicIndex.Enabled)
+
 	return exporterhelper.NewTracesExporter(
 		ctx,
 		set,
diff --git a/exporter/elasticsearchexporter/factory_test.go b/exporter/elasticsearchexporter/factory_test.go
index a6e2c356d981..5acc985e008e 100644
--- a/exporter/elasticsearchexporter/factory_test.go
+++ b/exporter/elasticsearchexporter/factory_test.go
@@ -35,15 +35,6 @@ func TestFactory_CreateLogsExporter(t *testing.T) {
 	require.NoError(t, exporter.Shutdown(context.Background()))
 }
 
-func TestFactory_CreateLogsExporter_Fail(t *testing.T) {
-	factory := NewFactory()
-	cfg := factory.CreateDefaultConfig()
-	params := exportertest.NewNopSettings()
-	_, err := factory.CreateLogsExporter(context.Background(), params, cfg)
-	require.Error(t, err, "expected an error when creating a logs exporter")
-	assert.EqualError(t, err, "cannot configure Elasticsearch exporter: exactly one of [endpoint, endpoints, cloudid] must be specified")
-}
-
 func TestFactory_CreateMetricsExporter(t *testing.T) {
 	factory := NewFactory()
 	cfg := withDefaultConfig(func(cfg *Config) {
@@ -70,15 +61,6 @@ func TestFactory_CreateTracesExporter(t *testing.T) {
 	require.NoError(t, exporter.Shutdown(context.Background()))
 }
 
-func TestFactory_CreateTracesExporter_Fail(t *testing.T) {
-	factory := NewFactory()
-	cfg := factory.CreateDefaultConfig()
-	params := exportertest.NewNopSettings()
-	_, err := factory.CreateTracesExporter(context.Background(), params, cfg)
-	require.Error(t, err, "expected an error when creating a traces exporter")
-	assert.EqualError(t, err, "cannot configure Elasticsearch exporter: exactly one of [endpoint, endpoints, cloudid] must be specified")
-}
-
 func TestFactory_CreateLogsAndTracesExporterWithDeprecatedIndexOption(t *testing.T) {
 	factory := NewFactory()
 	cfg := withDefaultConfig(func(cfg *Config) {

From 7e8ebd7b062275983fb2ca17e873a66c844d0103 Mon Sep 17 00:00:00 2001
From: Tyler Helmuth <12352919+TylerHelmuth@users.noreply.github.com>
Date: Mon, 16 Sep 2024 16:19:43 -0600
Subject: [PATCH 06/10] [pkg/ottl] Improve time performance (#35129)

**Description:**
Improves `Time` performance by move the conversion from our format to
Go's format to happen during startup.

Benchmarks before:

```
goos: darwin
goarch: arm64
pkg: github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/ottlfuncs
Benchmark_Time/simple_short_form-10         	1000000000	         0.0000079 ns/op	       0 B/op	       0 allocs/op
Benchmark_Time/simple_short_form_with_short_year_and_slashes-10         	1000000000	         0.0000115 ns/op	       0 B/op	       0 allocs/op
Benchmark_Time/month_day_year-10                                        	1000000000	         0.0000057 ns/op	       0 B/op	       0 allocs/op
Benchmark_Time/simple_long_form-10                                      	1000000000	         0.0000075 ns/op	       0 B/op	       0 allocs/op
Benchmark_Time/date_with_timestamp-10                                   	1000000000	         0.0000063 ns/op	       0 B/op	       0 allocs/op
Benchmark_Time/day_of_the_week_long_form-10                             	1000000000	         0.0000085 ns/op	       0 B/op	       0 allocs/op
Benchmark_Time/short_weekday,_short_month,_long_format-10               	1000000000	         0.0000089 ns/op	       0 B/op	       0 allocs/op
Benchmark_Time/short_months-10                                          	1000000000	         0.0000070 ns/op	       0 B/op	       0 allocs/op
Benchmark_Time/timestamp_with_time_zone_offset-10                       	1000000000	         0.0000665 ns/op	       0 B/op	       0 allocs/op
Benchmark_Time/short_date_with_timestamp_without_time_zone_offset-10    	1000000000	         0.0000428 ns/op	       0 B/op	       0 allocs/op
Benchmark_Time/RFC_3339_in_custom_format-10                             	1000000000	         0.0000345 ns/op	       0 B/op	       0 allocs/op
Benchmark_Time/RFC_3339_in_custom_format_before_2000-10                 	1000000000	         0.0000349 ns/op	       0 B/op	       0 allocs/op
Benchmark_Time/no_location-10                                           	1000000000	         0.0000035 ns/op	       0 B/op	       0 allocs/op
Benchmark_Time/with_location_-_America-10                               	1000000000	         0.0000104 ns/op	       0 B/op	       0 allocs/op
Benchmark_Time/with_location_-_Asia-10                                  	1000000000	         0.0000084 ns/op	       0 B/op	       0 allocs/op
Benchmark_Time/RFC_3339_in_custom_format_before_2000,_ignore_default_location-10         	1000000000	         0.0000379 ns/op	       0 B/op	       0 allocs/op
PASS
ok  	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/ottlfuncs	0.458s
```

Benchmark's after:

```
goos: darwin
goarch: arm64
pkg: github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/ottlfuncs
Benchmark_Time/simple_short_form-10         	1000000000	         0.0000054 ns/op	       0 B/op	       0 allocs/op
Benchmark_Time/simple_short_form_with_short_year_and_slashes-10         	1000000000	         0.0000037 ns/op	       0 B/op	       0 allocs/op
Benchmark_Time/month_day_year-10                                        	1000000000	         0.0000053 ns/op	       0 B/op	       0 allocs/op
Benchmark_Time/simple_long_form-10                                      	1000000000	         0.0000042 ns/op	       0 B/op	       0 allocs/op
Benchmark_Time/date_with_timestamp-10                                   	1000000000	         0.0000087 ns/op	       0 B/op	       0 allocs/op
Benchmark_Time/day_of_the_week_long_form-10                             	1000000000	         0.0000035 ns/op	       0 B/op	       0 allocs/op
Benchmark_Time/short_weekday,_short_month,_long_format-10               	1000000000	         0.0000036 ns/op	       0 B/op	       0 allocs/op
Benchmark_Time/short_months-10                                          	1000000000	         0.0000031 ns/op	       0 B/op	       0 allocs/op
Benchmark_Time/timestamp_with_time_zone_offset-10                       	1000000000	         0.0000491 ns/op	       0 B/op	       0 allocs/op
Benchmark_Time/short_date_with_timestamp_without_time_zone_offset-10    	1000000000	         0.0000381 ns/op	       0 B/op	       0 allocs/op
Benchmark_Time/RFC_3339_in_custom_format-10                             	1000000000	         0.0000365 ns/op	       0 B/op	       0 allocs/op
Benchmark_Time/RFC_3339_in_custom_format_before_2000-10                 	1000000000	         0.0000364 ns/op	       0 B/op	       0 allocs/op
Benchmark_Time/no_location-10                                           	1000000000	         0.0000028 ns/op	       0 B/op	       0 allocs/op
Benchmark_Time/with_location_-_America-10                               	1000000000	         0.0000017 ns/op	       0 B/op	       0 allocs/op
Benchmark_Time/with_location_-_Asia-10                                  	1000000000	         0.0000028 ns/op	       0 B/op	       0 allocs/op
Benchmark_Time/RFC_3339_in_custom_format_before_2000,_ignore_default_location-10         	1000000000	         0.0000393 ns/op	       0 B/op	       0 allocs/op
PASS
ok  	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/ottlfuncs	0.441s
```

**Link to tracking Issue:** <Issue number if applicable>
Closes
https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/35078

**Testing:** <Describe what testing was performed and which tests were
added.>
Added benchmark test
---
 pkg/ottl/ottlfuncs/func_time.go      |   8 +-
 pkg/ottl/ottlfuncs/func_time_test.go | 191 +++++++++++++++++++++++++++
 2 files changed, 197 insertions(+), 2 deletions(-)

diff --git a/pkg/ottl/ottlfuncs/func_time.go b/pkg/ottl/ottlfuncs/func_time.go
index b6d793cc3e5d..44371a94de6e 100644
--- a/pkg/ottl/ottlfuncs/func_time.go
+++ b/pkg/ottl/ottlfuncs/func_time.go
@@ -34,6 +34,11 @@ func Time[K any](inputTime ottl.StringGetter[K], format string, location ottl.Op
 	if format == "" {
 		return nil, fmt.Errorf("format cannot be nil")
 	}
+	gotimeFormat, err := timeutils.StrptimeToGotime(format)
+	if err != nil {
+		return nil, err
+	}
+
 	var defaultLocation *string
 	if !location.IsEmpty() {
 		l := location.Get()
@@ -44,7 +49,6 @@ func Time[K any](inputTime ottl.StringGetter[K], format string, location ottl.Op
 	if err != nil {
 		return nil, err
 	}
-
 	return func(ctx context.Context, tCtx K) (any, error) {
 		t, err := inputTime.Get(ctx, tCtx)
 		if err != nil {
@@ -53,7 +57,7 @@ func Time[K any](inputTime ottl.StringGetter[K], format string, location ottl.Op
 		if t == "" {
 			return nil, fmt.Errorf("time cannot be nil")
 		}
-		timestamp, err := timeutils.ParseStrptime(format, t, loc)
+		timestamp, err := timeutils.ParseGotime(gotimeFormat, t, loc)
 		if err != nil {
 			return nil, err
 		}
diff --git a/pkg/ottl/ottlfuncs/func_time_test.go b/pkg/ottl/ottlfuncs/func_time_test.go
index 41e62edaae04..cc9ce2a795f1 100644
--- a/pkg/ottl/ottlfuncs/func_time_test.go
+++ b/pkg/ottl/ottlfuncs/func_time_test.go
@@ -284,3 +284,194 @@ func Test_TimeFormatError(t *testing.T) {
 		})
 	}
 }
+
+func Benchmark_Time(t *testing.B) {
+	locationAmericaNewYork, _ := time.LoadLocation("America/New_York")
+	locationAsiaShanghai, _ := time.LoadLocation("Asia/Shanghai")
+
+	tests := []struct {
+		name     string
+		time     ottl.StringGetter[any]
+		format   string
+		expected time.Time
+		location string
+	}{
+		{
+			name: "simple short form",
+			time: &ottl.StandardStringGetter[any]{
+				Getter: func(_ context.Context, _ any) (any, error) {
+					return "2023-04-12", nil
+				},
+			},
+			format:   "%Y-%m-%d",
+			expected: time.Date(2023, 4, 12, 0, 0, 0, 0, time.Local),
+		},
+		{
+			name: "simple short form with short year and slashes",
+			time: &ottl.StandardStringGetter[any]{
+				Getter: func(_ context.Context, _ any) (any, error) {
+					return "11/11/11", nil
+				},
+			},
+			format:   "%d/%m/%y",
+			expected: time.Date(2011, 11, 11, 0, 0, 0, 0, time.Local),
+		},
+		{
+			name: "month day year",
+			time: &ottl.StandardStringGetter[any]{
+				Getter: func(_ context.Context, _ any) (any, error) {
+					return "02/04/2023", nil
+				},
+			},
+			format:   "%m/%d/%Y",
+			expected: time.Date(2023, 2, 4, 0, 0, 0, 0, time.Local),
+		},
+		{
+			name: "simple long form",
+			time: &ottl.StandardStringGetter[any]{
+				Getter: func(_ context.Context, _ any) (any, error) {
+					return "July 31, 1993", nil
+				},
+			},
+			format:   "%B %d, %Y",
+			expected: time.Date(1993, 7, 31, 0, 0, 0, 0, time.Local),
+		},
+		{
+			name: "date with timestamp",
+			time: &ottl.StandardStringGetter[any]{
+				Getter: func(_ context.Context, _ any) (any, error) {
+					return "Mar 14 2023 17:02:59", nil
+				},
+			},
+			format:   "%b %d %Y %H:%M:%S",
+			expected: time.Date(2023, 3, 14, 17, 02, 59, 0, time.Local),
+		},
+		{
+			name: "day of the week long form",
+			time: &ottl.StandardStringGetter[any]{
+				Getter: func(_ context.Context, _ any) (any, error) {
+					return "Monday, May 01, 2023", nil
+				},
+			},
+			format:   "%A, %B %d, %Y",
+			expected: time.Date(2023, 5, 1, 0, 0, 0, 0, time.Local),
+		},
+		{
+			name: "short weekday, short month, long format",
+			time: &ottl.StandardStringGetter[any]{
+				Getter: func(_ context.Context, _ any) (any, error) {
+					return "Sat, May 20, 2023", nil
+				},
+			},
+			format:   "%a, %b %d, %Y",
+			expected: time.Date(2023, 5, 20, 0, 0, 0, 0, time.Local),
+		},
+		{
+			name: "short months",
+			time: &ottl.StandardStringGetter[any]{
+				Getter: func(_ context.Context, _ any) (any, error) {
+					return "Feb 15, 2023", nil
+				},
+			},
+			format:   "%b %d, %Y",
+			expected: time.Date(2023, 2, 15, 0, 0, 0, 0, time.Local),
+		},
+		{
+			name: "timestamp with time zone offset",
+			time: &ottl.StandardStringGetter[any]{
+				Getter: func(_ context.Context, _ any) (any, error) {
+					return "2023-05-26 12:34:56 HST", nil
+				},
+			},
+			format:   "%Y-%m-%d %H:%M:%S %Z",
+			expected: time.Date(2023, 5, 26, 12, 34, 56, 0, time.FixedZone("HST", -10*60*60)),
+		},
+		{
+			name: "short date with timestamp without time zone offset",
+			time: &ottl.StandardStringGetter[any]{
+				Getter: func(_ context.Context, _ any) (any, error) {
+					return "2023-05-26T12:34:56 GMT", nil
+				},
+			},
+			format:   "%Y-%m-%dT%H:%M:%S %Z",
+			expected: time.Date(2023, 5, 26, 12, 34, 56, 0, time.FixedZone("GMT", 0)),
+		},
+		{
+			name: "RFC 3339 in custom format",
+			time: &ottl.StandardStringGetter[any]{
+				Getter: func(_ context.Context, _ any) (any, error) {
+					return "2012-11-01T22:08:41+0000 EST", nil
+				},
+			},
+			format:   "%Y-%m-%dT%H:%M:%S%z %Z",
+			expected: time.Date(2012, 11, 01, 22, 8, 41, 0, time.FixedZone("EST", 0)),
+		},
+		{
+			name: "RFC 3339 in custom format before 2000",
+			time: &ottl.StandardStringGetter[any]{
+				Getter: func(_ context.Context, _ any) (any, error) {
+					return "1986-10-01T00:17:33 MST", nil
+				},
+			},
+			format:   "%Y-%m-%dT%H:%M:%S %Z",
+			expected: time.Date(1986, 10, 01, 00, 17, 33, 00, time.FixedZone("MST", -7*60*60)),
+		},
+		{
+			name: "no location",
+			time: &ottl.StandardStringGetter[any]{
+				Getter: func(_ context.Context, _ any) (any, error) {
+					return "2022/01/01", nil
+				},
+			},
+			format:   "%Y/%m/%d",
+			expected: time.Date(2022, 01, 01, 0, 0, 0, 0, time.Local),
+		},
+		{
+			name: "with location - America",
+			time: &ottl.StandardStringGetter[any]{
+				Getter: func(_ context.Context, _ any) (any, error) {
+					return "2023-05-26 12:34:56", nil
+				},
+			},
+			format:   "%Y-%m-%d %H:%M:%S",
+			location: "America/New_York",
+			expected: time.Date(2023, 5, 26, 12, 34, 56, 0, locationAmericaNewYork),
+		},
+		{
+			name: "with location - Asia",
+			time: &ottl.StandardStringGetter[any]{
+				Getter: func(_ context.Context, _ any) (any, error) {
+					return "2023-05-26 12:34:56", nil
+				},
+			},
+			format:   "%Y-%m-%d %H:%M:%S",
+			location: "Asia/Shanghai",
+			expected: time.Date(2023, 5, 26, 12, 34, 56, 0, locationAsiaShanghai),
+		},
+		{
+			name: "RFC 3339 in custom format before 2000, ignore default location",
+			time: &ottl.StandardStringGetter[any]{
+				Getter: func(_ context.Context, _ any) (any, error) {
+					return "1986-10-01T00:17:33 MST", nil
+				},
+			},
+			location: "Asia/Shanghai",
+			format:   "%Y-%m-%dT%H:%M:%S %Z",
+			expected: time.Date(1986, 10, 01, 00, 17, 33, 00, time.FixedZone("MST", -7*60*60)),
+		},
+	}
+	for _, tt := range tests {
+		var locOptional ottl.Optional[string]
+		if tt.location != "" {
+			locOptional = ottl.NewTestingOptional(tt.location)
+		}
+		exprFunc, err := Time(tt.time, tt.format, locOptional)
+		assert.NoError(t, err)
+
+		t.Run(tt.name, func(t *testing.B) {
+			result, err := exprFunc(nil, nil)
+			assert.NoError(t, err)
+			assert.Equal(t, tt.expected.UnixNano(), result.(time.Time).UnixNano())
+		})
+	}
+}

From 0e945b6cb2dd9b23b8fe74ac9a50982e912eb253 Mon Sep 17 00:00:00 2001
From: Andrzej Stencel <andrzej.stencel@elastic.co>
Date: Tue, 17 Sep 2024 09:20:16 +0200
Subject: [PATCH 07/10] [exporter/file][chore][docs] Mention OTLP JSON File
 receiver in File exporter docs (#35197)

Users have trouble finding the receiver that can read the data written
by File exporter back into the collector (e.g. see this [CNCF Slack
post](https://cloud-native.slack.com/archives/C01N6P7KR6W/p1726222186384309)).
Let's mention the OTLP JSON File receiver in the File exporter's docs.
---
 exporter/fileexporter/README.md | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/exporter/fileexporter/README.md b/exporter/fileexporter/README.md
index ec550717b6d9..422e3cf6aeae 100644
--- a/exporter/fileexporter/README.md
+++ b/exporter/fileexporter/README.md
@@ -13,6 +13,10 @@
 [contrib]: https://github.com/open-telemetry/opentelemetry-collector-releases/tree/main/distributions/otelcol-contrib
 <!-- end autogenerated section -->
 
+Writes telemetry data to files on disk.
+
+Use the [OTLP JSON File receiver](../../receiver/otlpjsonfilereceiver/README.md) to read the data back into the collector (as long as the data was exported using OTLP JSON format).
+
 Exporter supports the following features：
 
 + Support for writing pipeline data to a file.

From c1e2df6b5a5b7397a168c8b46f72db26ec3dd3af Mon Sep 17 00:00:00 2001
From: Tyler Benson <734411+tylerbenson@users.noreply.github.com>
Date: Tue, 17 Sep 2024 03:43:48 -0400
Subject: [PATCH 08/10] [kafka] Add option to supply destination topic through
 context (#34503)

**Description:** <Describe what has changed.>
<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->

Add option to get destination topic from context. This allows for
upstream connectors to control the destination without polluting the
data being written.


**Link to tracking Issue:** <Issue number if applicable>
Fixes #34432

**Testing:** <Describe what testing was performed and which tests were
added.>
Added unit tests.

**Documentation:** <Describe the documentation added.>
Updated the component readme with the added setting.

---------

Co-authored-by: Pablo Baeyens <pablo.baeyens@datadoghq.com>
Co-authored-by: Sean Marciniak <30928402+MovieStoreGuy@users.noreply.github.com>
---
 .chloggen/kafka-topic-context.yaml            |  27 +++++
 .github/CODEOWNERS                            |   1 +
 .github/ISSUE_TEMPLATE/bug_report.yaml        |   1 +
 .github/ISSUE_TEMPLATE/feature_request.yaml   |   1 +
 .github/ISSUE_TEMPLATE/other.yaml             |   1 +
 .github/ISSUE_TEMPLATE/unmaintained.yaml      |   1 +
 cmd/otelcontribcol/builder-config.yaml        |   1 +
 cmd/otelcontribcol/go.mod                     |   3 +
 exporter/kafkaexporter/README.md              |  10 +-
 exporter/kafkaexporter/go.mod                 |   3 +
 exporter/kafkaexporter/kafka_exporter.go      |  32 +++---
 exporter/kafkaexporter/kafka_exporter_test.go | 105 +++++++++++++++++-
 pkg/kafka/topic/Makefile                      |   1 +
 pkg/kafka/topic/README.md                     |   4 +
 pkg/kafka/topic/go.mod                        |   3 +
 pkg/kafka/topic/kafka_ctx.go                  |  19 ++++
 pkg/kafka/topic/metadata.yaml                 |   6 +
 receiver/kafkareceiver/go.mod                 |   3 +
 versions.yaml                                 |   1 +
 19 files changed, 203 insertions(+), 20 deletions(-)
 create mode 100644 .chloggen/kafka-topic-context.yaml
 create mode 100644 pkg/kafka/topic/Makefile
 create mode 100644 pkg/kafka/topic/README.md
 create mode 100644 pkg/kafka/topic/go.mod
 create mode 100644 pkg/kafka/topic/kafka_ctx.go
 create mode 100644 pkg/kafka/topic/metadata.yaml

diff --git a/.chloggen/kafka-topic-context.yaml b/.chloggen/kafka-topic-context.yaml
new file mode 100644
index 000000000000..78d794be4efa
--- /dev/null
+++ b/.chloggen/kafka-topic-context.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: kafkaexporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add option to supply destination topic through context.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [34503, 34432]
+
+# (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: [api]
diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS
index d8b547da9834..56dab2a7b0ae 100644
--- a/.github/CODEOWNERS
+++ b/.github/CODEOWNERS
@@ -149,6 +149,7 @@ pkg/batchpersignal/                                                 @open-teleme
 pkg/datadog/                                                        @open-telemetry/collector-contrib-approvers @mx-psi @dineshg13 @liustanley @songy23 @mackjmr @ankitpatel96
 pkg/experimentalmetricmetadata/                                     @open-telemetry/collector-contrib-approvers @rmfitzpatrick
 pkg/golden/                                                         @open-telemetry/collector-contrib-approvers @djaglowski @atoulme
+pkg/kafka/topic/                                                    @open-telemetry/collector-contrib-approvers @pavolloffay @MovieStoreGuy
 pkg/ottl/                                                           @open-telemetry/collector-contrib-approvers @TylerHelmuth @kentquirk @bogdandrutu @evan-bradley
 pkg/pdatatest/                                                      @open-telemetry/collector-contrib-approvers @djaglowski @fatsheep9146
 pkg/pdatautil/                                                      @open-telemetry/collector-contrib-approvers @dmitryax
diff --git a/.github/ISSUE_TEMPLATE/bug_report.yaml b/.github/ISSUE_TEMPLATE/bug_report.yaml
index 9a3355295459..1ec802e83d93 100644
--- a/.github/ISSUE_TEMPLATE/bug_report.yaml
+++ b/.github/ISSUE_TEMPLATE/bug_report.yaml
@@ -145,6 +145,7 @@ body:
       - pkg/datadog
       - pkg/experimentalmetricmetadata
       - pkg/golden
+      - pkg/kafka/topic
       - pkg/ottl
       - pkg/pdatatest
       - pkg/pdatautil
diff --git a/.github/ISSUE_TEMPLATE/feature_request.yaml b/.github/ISSUE_TEMPLATE/feature_request.yaml
index 355b7f41489b..3027e60c72a8 100644
--- a/.github/ISSUE_TEMPLATE/feature_request.yaml
+++ b/.github/ISSUE_TEMPLATE/feature_request.yaml
@@ -139,6 +139,7 @@ body:
       - pkg/datadog
       - pkg/experimentalmetricmetadata
       - pkg/golden
+      - pkg/kafka/topic
       - pkg/ottl
       - pkg/pdatatest
       - pkg/pdatautil
diff --git a/.github/ISSUE_TEMPLATE/other.yaml b/.github/ISSUE_TEMPLATE/other.yaml
index 43d556c0b36e..e1e3cc57cd59 100644
--- a/.github/ISSUE_TEMPLATE/other.yaml
+++ b/.github/ISSUE_TEMPLATE/other.yaml
@@ -139,6 +139,7 @@ body:
       - pkg/datadog
       - pkg/experimentalmetricmetadata
       - pkg/golden
+      - pkg/kafka/topic
       - pkg/ottl
       - pkg/pdatatest
       - pkg/pdatautil
diff --git a/.github/ISSUE_TEMPLATE/unmaintained.yaml b/.github/ISSUE_TEMPLATE/unmaintained.yaml
index a5ceaea8a0fe..9f8c4ca17bab 100644
--- a/.github/ISSUE_TEMPLATE/unmaintained.yaml
+++ b/.github/ISSUE_TEMPLATE/unmaintained.yaml
@@ -144,6 +144,7 @@ body:
       - pkg/datadog
       - pkg/experimentalmetricmetadata
       - pkg/golden
+      - pkg/kafka/topic
       - pkg/ottl
       - pkg/pdatatest
       - pkg/pdatautil
diff --git a/cmd/otelcontribcol/builder-config.yaml b/cmd/otelcontribcol/builder-config.yaml
index 830cc3737be7..8a87ab0e5749 100644
--- a/cmd/otelcontribcol/builder-config.yaml
+++ b/cmd/otelcontribcol/builder-config.yaml
@@ -398,6 +398,7 @@ replaces:
   - github.com/open-telemetry/opentelemetry-collector-contrib/exporter/fileexporter => ../../exporter/fileexporter
   - github.com/open-telemetry/opentelemetry-collector-contrib/pkg/resourcetotelemetry => ../../pkg/resourcetotelemetry
   - github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden => ../../pkg/golden
+  - github.com/open-telemetry/opentelemetry-collector-contrib/pkg/kafka/topic  => ../../pkg/kafka/topic
   - github.com/open-telemetry/opentelemetry-collector-contrib/exporter/opencensusexporter => ../../exporter/opencensusexporter
   - github.com/open-telemetry/opentelemetry-collector-contrib/exporter/opensearchexporter => ../../exporter/opensearchexporter
   - github.com/open-telemetry/opentelemetry-collector-contrib/internal/metadataproviders => ../../internal/metadataproviders
diff --git a/cmd/otelcontribcol/go.mod b/cmd/otelcontribcol/go.mod
index dc06d555e700..7e8352d30598 100644
--- a/cmd/otelcontribcol/go.mod
+++ b/cmd/otelcontribcol/go.mod
@@ -658,6 +658,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/batchpersignal v0.109.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/datadog v0.0.0-00010101000000-000000000000 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/experimentalmetricmetadata v0.109.0 // indirect
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/kafka/topic v0.109.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl v0.109.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil v0.109.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/resourcetotelemetry v0.109.0 // indirect
@@ -1183,6 +1184,8 @@ replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/resourceto
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden => ../../pkg/golden
 
+replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/kafka/topic => ../../pkg/kafka/topic
+
 replace github.com/open-telemetry/opentelemetry-collector-contrib/exporter/opencensusexporter => ../../exporter/opencensusexporter
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/exporter/opensearchexporter => ../../exporter/opensearchexporter
diff --git a/exporter/kafkaexporter/README.md b/exporter/kafkaexporter/README.md
index c0ea3212cd9e..fc316e0494df 100644
--- a/exporter/kafkaexporter/README.md
+++ b/exporter/kafkaexporter/README.md
@@ -24,8 +24,8 @@ The following settings can be optionally configured:
 - `brokers` (default = localhost:9092): The list of kafka brokers.
 - `resolve_canonical_bootstrap_servers_only` (default = false): Whether to resolve then reverse-lookup broker IPs during startup.
 - `client_id` (default = "sarama"): The client ID to configure the Sarama Kafka client with. The client ID will be used for all produce requests.
-- `topic` (default = otlp_spans for traces, otlp_metrics for metrics, otlp_logs for logs): The name of the kafka topic to export to.
-- `topic_from_attribute` (default = ""): Specify the resource attribute whose value should be used as the message's topic. This option, when set, will take precedence over the default topic. If `topic_from_attribute` is not set, the message's topic will be set to the value of the configuration option `topic` instead. 
+- `topic` (default = otlp_spans for traces, otlp_metrics for metrics, otlp_logs for logs): The name of the default kafka topic to export to. See [Destination Topic](#destination-topic) below for more details.
+- `topic_from_attribute` (default = ""): Specify the resource attribute whose value should be used as the message's topic. See [Destination Topic](#destination-topic) below for more details. 
 - `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 or `ExportLogsServiceRequest` for logs.
   - `otlp_json`:  payload is JSON serialized from `ExportTraceServiceRequest` if set as a traces exporter or `ExportMetricsServiceRequest` for metrics or `ExportLogsServiceRequest` for logs. 
@@ -105,3 +105,9 @@ exporters:
       - localhost:9092
     protocol_version: 2.0.0
 ```
+
+## Destination Topic
+The destination topic can be defined in a few different ways and takes priority in the following order:
+1. When `topic_from_attribute` is configured, and the corresponding attribute is found on the ingested data, the value of this attribute is used.
+2. If a prior component in the collector pipeline sets the topic on the context via the `topic.WithTopic` function (from the `github.com/open-telemetry/opentelemetry-collector-contrib/pkg/kafka/topic` package), the value set in the context is used.
+3. Finally, the `topic` configuration is used as a default/fallback destination. 
diff --git a/exporter/kafkaexporter/go.mod b/exporter/kafkaexporter/go.mod
index ac5887ef94f7..049314db60e7 100644
--- a/exporter/kafkaexporter/go.mod
+++ b/exporter/kafkaexporter/go.mod
@@ -10,6 +10,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal v0.109.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/kafka v0.109.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/batchpersignal v0.109.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/kafka/topic v0.109.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil v0.109.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/jaeger v0.109.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/zipkin v0.109.0
@@ -106,6 +107,8 @@ replace github.com/open-telemetry/opentelemetry-collector-contrib/internal/kafka
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/batchpersignal => ../../pkg/batchpersignal
 
+replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/kafka/topic => ../../pkg/kafka/topic
+
 replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/jaeger => ../../pkg/translator/jaeger
 
 retract (
diff --git a/exporter/kafkaexporter/kafka_exporter.go b/exporter/kafkaexporter/kafka_exporter.go
index 845129b5b24e..fe3130cecc9c 100644
--- a/exporter/kafkaexporter/kafka_exporter.go
+++ b/exporter/kafkaexporter/kafka_exporter.go
@@ -19,6 +19,7 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/kafka"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/kafka/topic"
 )
 
 var errUnrecognizedEncoding = fmt.Errorf("unrecognized encoding")
@@ -40,8 +41,8 @@ func (ke kafkaErrors) Error() string {
 	return fmt.Sprintf("Failed to deliver %d messages due to %s", ke.count, ke.err)
 }
 
-func (e *kafkaTracesProducer) tracesPusher(_ context.Context, td ptrace.Traces) error {
-	messages, err := e.marshaler.Marshal(td, getTopic(&e.cfg, td.ResourceSpans()))
+func (e *kafkaTracesProducer) tracesPusher(ctx context.Context, td ptrace.Traces) error {
+	messages, err := e.marshaler.Marshal(td, getTopic(ctx, &e.cfg, td.ResourceSpans()))
 	if err != nil {
 		return consumererror.NewPermanent(err)
 	}
@@ -98,8 +99,8 @@ type kafkaMetricsProducer struct {
 	logger    *zap.Logger
 }
 
-func (e *kafkaMetricsProducer) metricsDataPusher(_ context.Context, md pmetric.Metrics) error {
-	messages, err := e.marshaler.Marshal(md, getTopic(&e.cfg, md.ResourceMetrics()))
+func (e *kafkaMetricsProducer) metricsDataPusher(ctx context.Context, md pmetric.Metrics) error {
+	messages, err := e.marshaler.Marshal(md, getTopic(ctx, &e.cfg, md.ResourceMetrics()))
 	if err != nil {
 		return consumererror.NewPermanent(err)
 	}
@@ -156,8 +157,8 @@ type kafkaLogsProducer struct {
 	logger    *zap.Logger
 }
 
-func (e *kafkaLogsProducer) logsDataPusher(_ context.Context, ld plog.Logs) error {
-	messages, err := e.marshaler.Marshal(ld, getTopic(&e.cfg, ld.ResourceLogs()))
+func (e *kafkaLogsProducer) logsDataPusher(ctx context.Context, ld plog.Logs) error {
+	messages, err := e.marshaler.Marshal(ld, getTopic(ctx, &e.cfg, ld.ResourceLogs()))
 	if err != nil {
 		return consumererror.NewPermanent(err)
 	}
@@ -283,16 +284,19 @@ type resource interface {
 	Resource() pcommon.Resource
 }
 
-func getTopic[T resource](cfg *Config, resources resourceSlice[T]) string {
-	if cfg.TopicFromAttribute == "" {
-		return cfg.Topic
-	}
-	for i := 0; i < resources.Len(); i++ {
-		rv, ok := resources.At(i).Resource().Attributes().Get(cfg.TopicFromAttribute)
-		if ok && rv.Str() != "" {
-			return rv.Str()
+func getTopic[T resource](ctx context.Context, cfg *Config, resources resourceSlice[T]) string {
+	if cfg.TopicFromAttribute != "" {
+		for i := 0; i < resources.Len(); i++ {
+			rv, ok := resources.At(i).Resource().Attributes().Get(cfg.TopicFromAttribute)
+			if ok && rv.Str() != "" {
+				return rv.Str()
+			}
 		}
 	}
+	contextTopic, ok := topic.FromContext(ctx)
+	if ok {
+		return contextTopic
+	}
 	return cfg.Topic
 }
 
diff --git a/exporter/kafkaexporter/kafka_exporter_test.go b/exporter/kafkaexporter/kafka_exporter_test.go
index 0edc6b9a52e5..670318887703 100644
--- a/exporter/kafkaexporter/kafka_exporter_test.go
+++ b/exporter/kafkaexporter/kafka_exporter_test.go
@@ -23,6 +23,7 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/kafka"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/kafka/topic"
 )
 
 func TestNewExporter_err_version(t *testing.T) {
@@ -205,6 +206,22 @@ func TestTracesPusher_attr(t *testing.T) {
 	require.NoError(t, err)
 }
 
+func TestTracesPusher_ctx(t *testing.T) {
+	c := sarama.NewConfig()
+	producer := mocks.NewSyncProducer(t, c)
+	producer.ExpectSendMessageAndSucceed()
+
+	p := kafkaTracesProducer{
+		producer:  producer,
+		marshaler: newPdataTracesMarshaler(&ptrace.ProtoMarshaler{}, defaultEncoding, false),
+	}
+	t.Cleanup(func() {
+		require.NoError(t, p.Close(context.Background()))
+	})
+	err := p.tracesPusher(topic.WithTopic(context.Background(), "my_topic"), testdata.GenerateTraces(2))
+	require.NoError(t, err)
+}
+
 func TestTracesPusher_err(t *testing.T) {
 	c := sarama.NewConfig()
 	producer := mocks.NewSyncProducer(t, c)
@@ -271,6 +288,22 @@ func TestMetricsDataPusher_attr(t *testing.T) {
 	require.NoError(t, err)
 }
 
+func TestMetricsDataPusher_ctx(t *testing.T) {
+	c := sarama.NewConfig()
+	producer := mocks.NewSyncProducer(t, c)
+	producer.ExpectSendMessageAndSucceed()
+
+	p := kafkaMetricsProducer{
+		producer:  producer,
+		marshaler: newPdataMetricsMarshaler(&pmetric.ProtoMarshaler{}, defaultEncoding, false),
+	}
+	t.Cleanup(func() {
+		require.NoError(t, p.Close(context.Background()))
+	})
+	err := p.metricsDataPusher(topic.WithTopic(context.Background(), "my_topic"), testdata.GenerateMetrics(2))
+	require.NoError(t, err)
+}
+
 func TestMetricsDataPusher_err(t *testing.T) {
 	c := sarama.NewConfig()
 	producer := mocks.NewSyncProducer(t, c)
@@ -337,6 +370,22 @@ func TestLogsDataPusher_attr(t *testing.T) {
 	require.NoError(t, err)
 }
 
+func TestLogsDataPusher_ctx(t *testing.T) {
+	c := sarama.NewConfig()
+	producer := mocks.NewSyncProducer(t, c)
+	producer.ExpectSendMessageAndSucceed()
+
+	p := kafkaLogsProducer{
+		producer:  producer,
+		marshaler: newPdataLogsMarshaler(&plog.ProtoMarshaler{}, defaultEncoding, false),
+	}
+	t.Cleanup(func() {
+		require.NoError(t, p.Close(context.Background()))
+	})
+	err := p.logsDataPusher(topic.WithTopic(context.Background(), "my_topic"), testdata.GenerateLogs(1))
+	require.NoError(t, err)
+}
+
 func TestLogsDataPusher_err(t *testing.T) {
 	c := sarama.NewConfig()
 	producer := mocks.NewSyncProducer(t, c)
@@ -410,6 +459,7 @@ func Test_GetTopic(t *testing.T) {
 	tests := []struct {
 		name      string
 		cfg       Config
+		ctx       context.Context
 		resource  any
 		wantTopic string
 	}{
@@ -419,6 +469,7 @@ func Test_GetTopic(t *testing.T) {
 				TopicFromAttribute: "resource-attr",
 				Topic:              "defaultTopic",
 			},
+			ctx:       topic.WithTopic(context.Background(), "context-topic"),
 			resource:  testdata.GenerateMetrics(1).ResourceMetrics(),
 			wantTopic: "resource-attr-val-1",
 		},
@@ -428,6 +479,7 @@ func Test_GetTopic(t *testing.T) {
 				TopicFromAttribute: "resource-attr",
 				Topic:              "defaultTopic",
 			},
+			ctx:       topic.WithTopic(context.Background(), "context-topic"),
 			resource:  testdata.GenerateTraces(1).ResourceSpans(),
 			wantTopic: "resource-attr-val-1",
 		},
@@ -437,6 +489,7 @@ func Test_GetTopic(t *testing.T) {
 				TopicFromAttribute: "resource-attr",
 				Topic:              "defaultTopic",
 			},
+			ctx:       topic.WithTopic(context.Background(), "context-topic"),
 			resource:  testdata.GenerateLogs(1).ResourceLogs(),
 			wantTopic: "resource-attr-val-1",
 		},
@@ -446,14 +499,58 @@ func Test_GetTopic(t *testing.T) {
 				TopicFromAttribute: "nonexistent_attribute",
 				Topic:              "defaultTopic",
 			},
+			ctx:       context.Background(),
+			resource:  testdata.GenerateMetrics(1).ResourceMetrics(),
+			wantTopic: "defaultTopic",
+		},
+
+		{
+			name: "Valid metric context, return topic name",
+			cfg: Config{
+				TopicFromAttribute: "nonexistent_attribute",
+				Topic:              "defaultTopic",
+			},
+			ctx:       topic.WithTopic(context.Background(), "context-topic"),
+			resource:  testdata.GenerateMetrics(1).ResourceMetrics(),
+			wantTopic: "context-topic",
+		},
+		{
+			name: "Valid trace context, return topic name",
+			cfg: Config{
+				TopicFromAttribute: "nonexistent_attribute",
+				Topic:              "defaultTopic",
+			},
+			ctx:       topic.WithTopic(context.Background(), "context-topic"),
+			resource:  testdata.GenerateTraces(1).ResourceSpans(),
+			wantTopic: "context-topic",
+		},
+		{
+			name: "Valid log context, return topic name",
+			cfg: Config{
+				TopicFromAttribute: "nonexistent_attribute",
+				Topic:              "defaultTopic",
+			},
+			ctx:       topic.WithTopic(context.Background(), "context-topic"),
+			resource:  testdata.GenerateLogs(1).ResourceLogs(),
+			wantTopic: "context-topic",
+		},
+
+		{
+			name: "Attribute not found",
+			cfg: Config{
+				TopicFromAttribute: "nonexistent_attribute",
+				Topic:              "defaultTopic",
+			},
+			ctx:       context.Background(),
 			resource:  testdata.GenerateMetrics(1).ResourceMetrics(),
 			wantTopic: "defaultTopic",
 		},
 		{
-			name: "TopicFromAttribute not set, return default topic",
+			name: "TopicFromAttribute, return default topic",
 			cfg: Config{
 				Topic: "defaultTopic",
 			},
+			ctx:       context.Background(),
 			resource:  testdata.GenerateMetrics(1).ResourceMetrics(),
 			wantTopic: "defaultTopic",
 		},
@@ -464,11 +561,11 @@ func Test_GetTopic(t *testing.T) {
 			topic := ""
 			switch r := tests[i].resource.(type) {
 			case pmetric.ResourceMetricsSlice:
-				topic = getTopic(&tests[i].cfg, r)
+				topic = getTopic(tests[i].ctx, &tests[i].cfg, r)
 			case ptrace.ResourceSpansSlice:
-				topic = getTopic(&tests[i].cfg, r)
+				topic = getTopic(tests[i].ctx, &tests[i].cfg, r)
 			case plog.ResourceLogsSlice:
-				topic = getTopic(&tests[i].cfg, r)
+				topic = getTopic(tests[i].ctx, &tests[i].cfg, r)
 			}
 			assert.Equal(t, tests[i].wantTopic, topic)
 		})
diff --git a/pkg/kafka/topic/Makefile b/pkg/kafka/topic/Makefile
new file mode 100644
index 000000000000..bdd863a203be
--- /dev/null
+++ b/pkg/kafka/topic/Makefile
@@ -0,0 +1 @@
+include ../../../Makefile.Common
diff --git a/pkg/kafka/topic/README.md b/pkg/kafka/topic/README.md
new file mode 100644
index 000000000000..663d969e5d61
--- /dev/null
+++ b/pkg/kafka/topic/README.md
@@ -0,0 +1,4 @@
+# Kafka Topic Context Accessor
+
+This module is used for accessing the topic within a context.
+See the [kafka exporter readme](../../../exporter/kafkaexporter/README.md#destination-topic) for more details.
diff --git a/pkg/kafka/topic/go.mod b/pkg/kafka/topic/go.mod
new file mode 100644
index 000000000000..5c27986a6d6e
--- /dev/null
+++ b/pkg/kafka/topic/go.mod
@@ -0,0 +1,3 @@
+module github.com/open-telemetry/opentelemetry-collector-contrib/pkg/kafka/topic
+
+go 1.22
diff --git a/pkg/kafka/topic/kafka_ctx.go b/pkg/kafka/topic/kafka_ctx.go
new file mode 100644
index 000000000000..603863615fb4
--- /dev/null
+++ b/pkg/kafka/topic/kafka_ctx.go
@@ -0,0 +1,19 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package topic // import "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/kafka/topic"
+
+import (
+	"context"
+)
+
+func WithTopic(ctx context.Context, topic string) context.Context {
+	return context.WithValue(ctx, topicContextKey{}, topic)
+}
+
+func FromContext(ctx context.Context) (string, bool) {
+	contextTopic, ok := ctx.Value(topicContextKey{}).(string)
+	return contextTopic, ok
+}
+
+type topicContextKey struct{}
diff --git a/pkg/kafka/topic/metadata.yaml b/pkg/kafka/topic/metadata.yaml
new file mode 100644
index 000000000000..708ccce63f83
--- /dev/null
+++ b/pkg/kafka/topic/metadata.yaml
@@ -0,0 +1,6 @@
+type: topic
+
+status:
+  class: pkg
+  codeowners:
+    active: [pavolloffay, MovieStoreGuy]
diff --git a/receiver/kafkareceiver/go.mod b/receiver/kafkareceiver/go.mod
index 8decd129de0a..55b676e3ccf0 100644
--- a/receiver/kafkareceiver/go.mod
+++ b/receiver/kafkareceiver/go.mod
@@ -69,6 +69,7 @@ require (
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/batchpersignal v0.109.0 // indirect
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/kafka/topic v0.109.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil v0.109.0 // indirect
 	github.com/pierrec/lz4/v4 v4.1.21 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
@@ -122,6 +123,8 @@ retract (
 	v0.65.0
 )
 
+replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/kafka/topic => ../../pkg/kafka/topic
+
 replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil => ../../pkg/pdatautil
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest => ../../pkg/pdatatest
diff --git a/versions.yaml b/versions.yaml
index 27479bf26abd..c63af9fe4b11 100644
--- a/versions.yaml
+++ b/versions.yaml
@@ -142,6 +142,7 @@ module-sets:
       - github.com/open-telemetry/opentelemetry-collector-contrib/pkg/batchpersignal
       - github.com/open-telemetry/opentelemetry-collector-contrib/pkg/experimentalmetricmetadata
       - github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden
+      - github.com/open-telemetry/opentelemetry-collector-contrib/pkg/kafka/topic
       - github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl
       - github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest
       - github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil

From 624b3682b96067aca40a0b70beb844e94f561a50 Mon Sep 17 00:00:00 2001
From: Florian Bacher <florian.bacher@dynatrace.com>
Date: Tue, 17 Sep 2024 12:12:47 +0200
Subject: [PATCH 09/10] [chore] adapt message sent by the syslog data sender to
 fit rfc5424 format (#35201)

**Description:** This PR adapts the syslog data sender to generate valid
messages adhering to the RFC5424 format. Previously, the data sender
generated invalid messages causing the syslog message parser to fail
(due to empty attribute values, a missing version number and SDID

**Link to tracking Issue:** -

**Testing:** -

**Documentation:** -

---------

Signed-off-by: Florian Bacher <florian.bacher@dynatrace.com>
---
 testbed/datasenders/syslog.go | 14 ++++++++++----
 1 file changed, 10 insertions(+), 4 deletions(-)

diff --git a/testbed/datasenders/syslog.go b/testbed/datasenders/syslog.go
index 3417d9bbc390..24cec672a6c7 100644
--- a/testbed/datasenders/syslog.go
+++ b/testbed/datasenders/syslog.go
@@ -91,14 +91,20 @@ func (f *SyslogWriter) GenConfigYAMLStr() string {
 func (f *SyslogWriter) Send(lr plog.LogRecord) error {
 	ts := time.Unix(int64(lr.Timestamp()/1_000_000_000), int64(lr.Timestamp()%1_000_000_000)).Format(time.RFC3339Nano)
 	sdid := strings.Builder{}
-	sdid.WriteString(fmt.Sprintf("%s=\"%s\" ", "trace_id", lr.TraceID()))
-	sdid.WriteString(fmt.Sprintf("%s=\"%s\" ", "span_id", lr.SpanID()))
+	if lr.TraceID().String() != "" {
+		sdid.WriteString(fmt.Sprintf("%s=\"%s\" ", "trace_id", lr.TraceID()))
+	}
+	if lr.SpanID().String() != "" {
+		sdid.WriteString(fmt.Sprintf("%s=\"%s\" ", "span_id", lr.SpanID()))
+	}
 	sdid.WriteString(fmt.Sprintf("%s=\"%d\" ", "trace_flags", lr.Flags()))
 	lr.Attributes().Range(func(k string, v pcommon.Value) bool {
-		sdid.WriteString(fmt.Sprintf("%s=\"%s\" ", k, v.Str()))
+		if v.Str() != "" {
+			sdid.WriteString(fmt.Sprintf("%s=\"%s\" ", k, v.Str()))
+		}
 		return true
 	})
-	msg := fmt.Sprintf("<166> %s 127.0.0.1 - - - [%s] %s\n", ts, sdid.String(), lr.Body().Str())
+	msg := fmt.Sprintf("<166>1 %s 127.0.0.1 - - - [test@12345 %s] %s\n", ts, strings.TrimSpace(sdid.String()), lr.Body().Str())
 
 	f.buf = append(f.buf, msg)
 	return f.SendCheck()

From 3f4dbf17e9cee51b4bcd5dea2e6409dfd66a456b Mon Sep 17 00:00:00 2001
From: Alex Boten <223565+codeboten@users.noreply.github.com>
Date: Tue, 17 Sep 2024 08:19:19 -0700
Subject: [PATCH 10/10] [processor/groupbytrace] ensure metric has unit
 (#35221)

A missing unit will break validation when mdatagen is updated

---------

Signed-off-by: Alex Boten <223565+codeboten@users.noreply.github.com>
---
 .chloggen/codeboten_add-unit.yaml             | 27 +++++++++++++++++++
 .../groupbytraceprocessor/documentation.md    |  2 +-
 .../internal/metadata/generated_telemetry.go  |  2 +-
 processor/groupbytraceprocessor/metadata.yaml |  1 +
 4 files changed, 30 insertions(+), 2 deletions(-)
 create mode 100644 .chloggen/codeboten_add-unit.yaml

diff --git a/.chloggen/codeboten_add-unit.yaml b/.chloggen/codeboten_add-unit.yaml
new file mode 100644
index 000000000000..53b088706a69
--- /dev/null
+++ b/.chloggen/codeboten_add-unit.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: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: groupbytraceprocessor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Ensure processor_groupbytrace_incomplete_releases metric has a unit.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [35221]
+
+# (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: []
diff --git a/processor/groupbytraceprocessor/documentation.md b/processor/groupbytraceprocessor/documentation.md
index 384939278508..be8bb8183fe8 100644
--- a/processor/groupbytraceprocessor/documentation.md
+++ b/processor/groupbytraceprocessor/documentation.md
@@ -28,7 +28,7 @@ Releases that are suspected to have been incomplete
 
 | Unit | Metric Type | Value Type | Monotonic |
 | ---- | ----------- | ---------- | --------- |
-| <nil> | Sum | Int | true |
+| {releases} | Sum | Int | true |
 
 ### otelcol_processor_groupbytrace_num_events_in_queue
 
diff --git a/processor/groupbytraceprocessor/internal/metadata/generated_telemetry.go b/processor/groupbytraceprocessor/internal/metadata/generated_telemetry.go
index b2f4db20c319..58fe789590e6 100644
--- a/processor/groupbytraceprocessor/internal/metadata/generated_telemetry.go
+++ b/processor/groupbytraceprocessor/internal/metadata/generated_telemetry.go
@@ -67,7 +67,7 @@ func NewTelemetryBuilder(settings component.TelemetrySettings, options ...teleme
 	builder.ProcessorGroupbytraceIncompleteReleases, err = builder.meters[configtelemetry.LevelBasic].Int64Counter(
 		"otelcol_processor_groupbytrace_incomplete_releases",
 		metric.WithDescription("Releases that are suspected to have been incomplete"),
-		metric.WithUnit("<nil>"),
+		metric.WithUnit("{releases}"),
 	)
 	errs = errors.Join(errs, err)
 	builder.ProcessorGroupbytraceNumEventsInQueue, err = builder.meters[configtelemetry.LevelBasic].Int64Gauge(
diff --git a/processor/groupbytraceprocessor/metadata.yaml b/processor/groupbytraceprocessor/metadata.yaml
index ef574c02edbf..0e2fab550e12 100644
--- a/processor/groupbytraceprocessor/metadata.yaml
+++ b/processor/groupbytraceprocessor/metadata.yaml
@@ -57,6 +57,7 @@ telemetry:
     processor_groupbytrace_incomplete_releases:
       enabled: true
       description: Releases that are suspected to have been incomplete
+      unit: "{releases}"
       sum:
         value_type: int
         monotonic: true