[chore] Move queue-batch from internal to internal/queuebatch (#12698)

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

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

<!--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.-->

Signed-off-by: Bogdan Drutu <bogdandrutu@gmail.com>

Author: bogdandrutu

exporter/exporterhelper/internal/base_exporter_test.go

@@ -16,7 +16,6 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/config/configretry"
-	"go.opentelemetry.io/collector/exporter"
 	"go.opentelemetry.io/collector/exporter/exporterbatcher"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
@@ -26,19 +25,8 @@ import (
 	"go.opentelemetry.io/collector/pipeline"
 )
 
-var (
-	defaultType     = component.MustNewType("test")
-	defaultSignal   = pipeline.SignalMetrics
-	defaultID       = component.NewID(defaultType)
-	defaultSettings = func() exporter.Settings {
-		set := exportertest.NewNopSettings(exportertest.NopType)
-		set.ID = defaultID
-		return set
-	}()
-)
-
 func TestBaseExporter(t *testing.T) {
-	be, err := NewBaseExporter(defaultSettings, defaultSignal, noopExport)
+	be, err := NewBaseExporter(exportertest.NewNopSettings(exportertest.NopType), pipeline.SignalMetrics, noopExport)
 	require.NoError(t, err)
 	require.NoError(t, be.Start(context.Background(), componenttest.NewNopHost()))
 	require.NoError(t, be.Shutdown(context.Background()))
@@ -47,7 +35,7 @@ func TestBaseExporter(t *testing.T) {
 func TestBaseExporterWithOptions(t *testing.T) {
 	want := errors.New("my error")
 	be, err := NewBaseExporter(
-		defaultSettings, defaultSignal, noopExport,
+		exportertest.NewNopSettings(exportertest.NopType), pipeline.SignalMetrics, noopExport,
 		WithStart(func(context.Context, component.Host) error { return want }),
 		WithShutdown(func(context.Context) error { return want }),
 		WithTimeout(NewDefaultTimeoutConfig()),
@@ -58,18 +46,18 @@ func TestBaseExporterWithOptions(t *testing.T) {
 }
 
 func TestQueueOptionsWithRequestExporter(t *testing.T) {
-	bs, err := NewBaseExporter(exportertest.NewNopSettings(exportertest.NopType), defaultSignal, noopExport,
+	bs, err := NewBaseExporter(exportertest.NewNopSettings(exportertest.NopType), pipeline.SignalMetrics, noopExport,
 		WithRetry(configretry.NewDefaultBackOffConfig()))
 	require.NoError(t, err)
 	require.Nil(t, bs.queueBatchSettings.Encoding)
-	_, err = NewBaseExporter(exportertest.NewNopSettings(exportertest.NopType), defaultSignal, noopExport,
+	_, err = NewBaseExporter(exportertest.NewNopSettings(exportertest.NopType), pipeline.SignalMetrics, noopExport,
 		WithRetry(configretry.NewDefaultBackOffConfig()), WithQueue(exporterqueue.NewDefaultConfig()))
 	require.Error(t, err)
 
 	qCfg := exporterqueue.NewDefaultConfig()
 	storageID := component.NewID(component.MustNewType("test"))
 	qCfg.StorageID = &storageID
-	_, err = NewBaseExporter(exportertest.NewNopSettings(exportertest.NopType), defaultSignal, noopExport,
+	_, err = NewBaseExporter(exportertest.NewNopSettings(exportertest.NopType), pipeline.SignalMetrics, noopExport,
 		WithQueueBatchSettings(newFakeQueueBatch(&requesttest.FakeRequest{Items: 1})),
 		WithRetry(configretry.NewDefaultBackOffConfig()),
 		WithQueueBatch(qCfg, queuebatch.Settings[request.Request]{}))
@@ -84,7 +72,7 @@ func TestBaseExporterLogging(t *testing.T) {
 	rCfg.Enabled = false
 	qCfg := exporterqueue.NewDefaultConfig()
 	qCfg.Enabled = false
-	bs, err := NewBaseExporter(set, defaultSignal, errExport,
+	bs, err := NewBaseExporter(set, pipeline.SignalMetrics, errExport,
 		WithQueueBatch(qCfg, newFakeQueueBatch(&requesttest.FakeRequest{})),
 		WithBatcher(exporterbatcher.NewDefaultConfig()),
 		WithRetry(rCfg))
@@ -167,3 +155,15 @@ func noopExport(context.Context, request.Request) error {
 func newFakeQueueBatch(mr request.Request) queuebatch.Settings[request.Request] {
 	return queuebatch.Settings[request.Request]{Encoding: &fakeEncoding{mr: mr}}
 }
+
+type fakeEncoding struct {
+	mr request.Request
+}
+
+func (f fakeEncoding) Marshal(request.Request) ([]byte, error) {
+	return []byte("mockRequest"), nil
+}
+
+func (f fakeEncoding) Unmarshal([]byte) (request.Request, error) {
+	return f.mr, nil
+}

exporter/exporterhelper/internal/queue_sender.go

@@ -5,11 +5,9 @@ package internal // import "go.opentelemetry.io/collector/exporter/exporterhelpe
 
 import (
 	"context"
-	"errors"
 
 	"go.uber.org/zap"
 
-	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/exporter/exporterbatcher"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
@@ -35,57 +33,6 @@ func NewQueueSender(
 		}
 		return nil
 	}
-	return NewQueueBatch(qSet, qCfg, bCfg, exportFunc)
-}
-
-type QueueBatch struct {
-	queue   queuebatch.Queue[request.Request]
-	batcher queuebatch.Batcher[request.Request]
-}
-
-func NewQueueBatch(
-	qSet queuebatch.QueueSettings[request.Request],
-	qCfg exporterqueue.Config,
-	bCfg exporterbatcher.Config,
-	next sender.SendFunc[request.Request],
-) (*QueueBatch, error) {
-	b, err := queuebatch.NewBatcher(bCfg, next, qCfg.NumConsumers)
-	if err != nil {
-		return nil, err
-	}
-	// TODO: https://github.com/open-telemetry/opentelemetry-collector/issues/12244
-	if bCfg.Enabled {
-		qCfg.NumConsumers = 1
-	}
-	q, err := queuebatch.NewQueue(context.Background(), qSet, qCfg, b.Consume)
-	if err != nil {
-		return nil, err
-	}
-	oq, err := newObsQueue(qSet, q)
-	if err != nil {
-		return nil, err
-	}
-
-	return &QueueBatch{queue: oq, batcher: b}, nil
-}
-
-// Start is invoked during service startup.
-func (qs *QueueBatch) Start(ctx context.Context, host component.Host) error {
-	if err := qs.queue.Start(ctx, host); err != nil {
-		return err
-	}
-
-	return qs.batcher.Start(ctx, host)
-}
-
-// Shutdown is invoked during service shutdown.
-func (qs *QueueBatch) Shutdown(ctx context.Context) error {
-	// Stop the queue and batcher, this will drain the queue and will call the retry (which is stopped) that will only
-	// try once every request.
-	return errors.Join(qs.queue.Shutdown(ctx), qs.batcher.Shutdown(ctx))
-}
 
-// Send implements the requestSender interface. It puts the request in the queue.
-func (qs *QueueBatch) Send(ctx context.Context, req request.Request) error {
-	return qs.queue.Offer(ctx, req)
+	return queuebatch.NewQueueBatch(qSet, qCfg, bCfg, exportFunc)
 }