[exporterhelper][queuebatcher] partitioner as the interface to get partition key (#12803)

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

This RP implements a type `partitioner` that serves as the interface to
get partition key.

<!-- Issue number if applicable -->
#### Link to tracking issue
#12795

<!--Describe what testing was performed and which tests were added.-->
#### Testing

<!--Describe the documentation added.-->
#### Documentation

<!--Please delete paragraphs that you did not use before submitting.-->

Author: sfc-gh-sili

exporter/debugexporter/go.mod

@@ -124,3 +124,5 @@ replace go.opentelemetry.io/collector/featuregate => ../../featuregate
 replace go.opentelemetry.io/collector/extension/xextension => ../../extension/xextension
 
 replace go.opentelemetry.io/collector/internal/telemetry => ../../internal/telemetry
+
+replace go.opentelemetry.io/collector/client => ../../client

exporter/exporterhelper/internal/queue_sender.go

@@ -20,8 +20,9 @@ import (
 
 // QueueBatchSettings is a subset of the queuebatch.Settings that are needed when used within an Exporter.
 type QueueBatchSettings[T any] struct {
-	Encoding queuebatch.Encoding[T]
-	Sizers   map[request.SizerType]request.Sizer[T]
+	Encoding    queuebatch.Encoding[T]
+	Sizers      map[request.SizerType]request.Sizer[T]
+	Partitioner queuebatch.Partitioner[T]
 }
 
 // NewDefaultQueueConfig returns the default config for queuebatch.Config.

exporter/exporterhelper/internal/queuebatch/partitioner.go

@@ -0,0 +1,33 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package queuebatch // import "go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
+
+import (
+	"context"
+
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
+)
+
+// Partitioner is an interface that returns the the partition key of the given element.
+type Partitioner[T any] interface {
+	GetKey(context.Context, T) string
+}
+
+type GetKeyFunc[T any] func(context.Context, T) string
+
+func (f GetKeyFunc[T]) GetKey(ctx context.Context, t T) string {
+	return f(ctx, t)
+}
+
+type basePartitioner struct {
+	GetKeyFunc[request.Request]
+}
+
+func NewPartitioner(
+	getKeyFunc GetKeyFunc[request.Request],
+) Partitioner[request.Request] {
+	return &basePartitioner{
+		GetKeyFunc: getKeyFunc,
+	}
+}

exporter/exporterhelper/internal/queuebatch/partitioner_test.go

@@ -0,0 +1,42 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package queuebatch
+
+import (
+	"context"
+	"strconv"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"go.opentelemetry.io/collector/client"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/requesttest"
+)
+
+func TestPartitioner_GetKeyFromRequest(t *testing.T) {
+	partitioner := NewPartitioner(func(_ context.Context, req request.Request) string {
+		return strconv.Itoa(req.(*requesttest.FakeRequest).ItemsCount())
+	})
+
+	require.Equal(t, "2", partitioner.GetKey(context.Background(), &requesttest.FakeRequest{Items: 2}))
+	require.Equal(t, "3", partitioner.GetKey(context.Background(), &requesttest.FakeRequest{Items: 3}))
+	require.Equal(t, "4", partitioner.GetKey(context.Background(), &requesttest.FakeRequest{Items: 4}))
+}
+
+func TestPartitioner_GetKeyFromContext(t *testing.T) {
+	partitioner := NewPartitioner(func(ctx context.Context, _ request.Request) string {
+		return client.FromContext(ctx).Metadata.Get("metadata_key")[0]
+	})
+
+	ctx1 := client.NewContext(context.Background(), client.Info{
+		Metadata: client.NewMetadata(map[string][]string{"metadata_key": {"partition1"}}),
+	})
+	require.Equal(t, "partition1", partitioner.GetKey(ctx1, &requesttest.FakeRequest{Items: 2}))
+
+	ctx2 := client.NewContext(context.Background(), client.Info{
+		Metadata: client.NewMetadata(map[string][]string{"metadata_key": {"partition2"}}),
+	})
+	require.Equal(t, "partition2", partitioner.GetKey(ctx2, &requesttest.FakeRequest{Items: 2}))
+}

exporter/exporterhelper/xexporterhelper/go.mod

@@ -117,3 +117,5 @@ replace go.opentelemetry.io/collector/featuregate => ../../../featuregate
 replace go.opentelemetry.io/collector/confmap => ../../../confmap
 
 replace go.opentelemetry.io/collector/internal/telemetry => ../../../internal/telemetry
+
+replace go.opentelemetry.io/collector/client => ../../../client

exporter/exportertest/go.mod

@@ -107,3 +107,5 @@ replace go.opentelemetry.io/collector/featuregate => ../../featuregate
 replace go.opentelemetry.io/collector/confmap => ../../confmap
 
 replace go.opentelemetry.io/collector/internal/telemetry => ../../internal/telemetry
+
+replace go.opentelemetry.io/collector/client => ../../client

exporter/go.mod

@@ -5,6 +5,7 @@ go 1.23.0
 require (
 	github.com/cenkalti/backoff/v5 v5.0.2
 	github.com/stretchr/testify v1.10.0
+	go.opentelemetry.io/collector/client v1.29.0
 	go.opentelemetry.io/collector/component v1.30.0
 	go.opentelemetry.io/collector/component/componenttest v0.124.0
 	go.opentelemetry.io/collector/config/configretry v1.30.0
@@ -112,3 +113,5 @@ replace go.opentelemetry.io/collector/featuregate => ../featuregate
 replace go.opentelemetry.io/collector/confmap => ../confmap
 
 replace go.opentelemetry.io/collector/internal/telemetry => ../internal/telemetry
+
+replace go.opentelemetry.io/collector/client => ../client

exporter/nopexporter/go.mod

@@ -107,3 +107,5 @@ replace go.opentelemetry.io/collector/featuregate => ../../featuregate
 replace go.opentelemetry.io/collector/extension/xextension => ../../extension/xextension
 
 replace go.opentelemetry.io/collector/internal/telemetry => ../../internal/telemetry
+
+replace go.opentelemetry.io/collector/client => ../../client

exporter/xexporter/go.mod

@@ -88,3 +88,5 @@ replace go.opentelemetry.io/collector/featuregate => ../../featuregate
 replace go.opentelemetry.io/collector/confmap => ../../confmap
 
 replace go.opentelemetry.io/collector/internal/telemetry => ../../internal/telemetry
+
+replace go.opentelemetry.io/collector/client => ../../client