[exporter][chore] Change queue batcher to use exportFunc instead of r…

…equest.export() (#11636)

#### Description

This PR changes queue batcher to use `exportFunc` instead of
`request.export()`. This makes testing easier and avoid passing
unnecessary detail to the exporter batcher.

#### Link to tracking issue

#8122
#10368

exporter/internal/queue/batcher.go

@@ -28,16 +28,21 @@ type BaseBatcher struct {
 	queue      Queue[internal.Request]
 	maxWorkers int
 	workerPool chan bool
+	exportFunc func(ctx context.Context, req internal.Request) error
 	stopWG     sync.WaitGroup
 }
 
-func NewBatcher(batchCfg exporterbatcher.Config, queue Queue[internal.Request], maxWorkers int) (Batcher, error) {
+func NewBatcher(batchCfg exporterbatcher.Config,
+	queue Queue[internal.Request],
+	exportFunc func(ctx context.Context, req internal.Request) error,
+	maxWorkers int) (Batcher, error) {
 	if !batchCfg.Enabled {
 		return &DisabledBatcher{
 			BaseBatcher{
 				batchCfg:   batchCfg,
 				queue:      queue,
 				maxWorkers: maxWorkers,
+				exportFunc: exportFunc,
 				stopWG:     sync.WaitGroup{},
 			},
 		}, nil
@@ -48,6 +53,7 @@ func NewBatcher(batchCfg exporterbatcher.Config, queue Queue[internal.Request],
 			batchCfg:   batchCfg,
 			queue:      queue,
 			maxWorkers: maxWorkers,
+			exportFunc: exportFunc,
 			stopWG:     sync.WaitGroup{},
 		},
 	}, nil
@@ -65,7 +71,7 @@ func (qb *BaseBatcher) startWorkerPool() {
 
 // flush exports the incoming batch synchronously.
 func (qb *BaseBatcher) flush(batchToFlush batch) {
-	err := batchToFlush.req.Export(batchToFlush.ctx)
+	err := qb.exportFunc(batchToFlush.ctx, batchToFlush.req)
 	for _, idx := range batchToFlush.idxList {
 		qb.queue.OnProcessingFinished(idx, err)
 	}

exporter/internal/queue/default_batcher_test.go

@@ -50,7 +50,9 @@ func TestDefaultBatcher_NoSplit_MinThresholdZero_TimeoutDisabled(t *testing.T) {
 					Capacity: 10,
 				})
 
-			ba, err := NewBatcher(cfg, q, tt.maxWorkers)
+			ba, err := NewBatcher(cfg, q,
+				func(ctx context.Context, req internal.Request) error { return req.Export(ctx) },
+				tt.maxWorkers)
 			require.NoError(t, err)
 
 			require.NoError(t, q.Start(context.Background(), componenttest.NewNopHost()))
@@ -108,7 +110,9 @@ func TestDefaultBatcher_NoSplit_TimeoutDisabled(t *testing.T) {
 					Capacity: 10,
 				})
 
-			ba, err := NewBatcher(cfg, q, tt.maxWorkers)
+			ba, err := NewBatcher(cfg, q,
+				func(ctx context.Context, req internal.Request) error { return req.Export(ctx) },
+				tt.maxWorkers)
 			require.NoError(t, err)
 
 			require.NoError(t, q.Start(context.Background(), componenttest.NewNopHost()))
@@ -172,7 +176,9 @@ func TestDefaultBatcher_NoSplit_WithTimeout(t *testing.T) {
 					Capacity: 10,
 				})
 
-			ba, err := NewBatcher(cfg, q, tt.maxWorkers)
+			ba, err := NewBatcher(cfg, q,
+				func(ctx context.Context, req internal.Request) error { return req.Export(ctx) },
+				tt.maxWorkers)
 			require.NoError(t, err)
 
 			require.NoError(t, q.Start(context.Background(), componenttest.NewNopHost()))
@@ -236,7 +242,9 @@ func TestDefaultBatcher_Split_TimeoutDisabled(t *testing.T) {
 					Capacity: 10,
 				})
 
-			ba, err := NewBatcher(cfg, q, tt.maxWorkers)
+			ba, err := NewBatcher(cfg, q,
+				func(ctx context.Context, req internal.Request) error { return req.Export(ctx) },
+				tt.maxWorkers)
 			require.NoError(t, err)
 
 			require.NoError(t, q.Start(context.Background(), componenttest.NewNopHost()))

exporter/internal/queue/disabled_batcher_test.go

@@ -46,7 +46,9 @@ func TestDisabledBatcher_Basic(t *testing.T) {
 					Capacity: 10,
 				})
 
-			ba, err := NewBatcher(cfg, q, tt.maxWorkers)
+			ba, err := NewBatcher(cfg, q,
+				func(ctx context.Context, req internal.Request) error { return req.Export(ctx) },
+				tt.maxWorkers)
 			require.NoError(t, err)
 
 			require.NoError(t, q.Start(context.Background(), componenttest.NewNopHost()))