Remove old push based batcher

Signed-off-by: Bogdan Drutu <[email protected]>

.chloggen/rm-old-push-based-batcher.yaml

@@ -0,0 +1,25 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: deprecation
+
+# The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver)
+component: exporterhelerp
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Deprecate alpha featuregate "exporter.UsePullingBasedExporterQueueBatcher". Functionality is enabled by default.
+
+# One or more tracking issues or pull requests related to the change
+issues: [12233]
+
+# (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:
+
+# 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/exporterhelper/internal/base_exporter.go

@@ -25,11 +25,12 @@ import (
 	"go.opentelemetry.io/collector/pipeline"
 )
 
-var usePullingBasedExporterQueueBatcher = featuregate.GlobalRegistry().MustRegister(
+var _ = featuregate.GlobalRegistry().MustRegister(
 	"exporter.UsePullingBasedExporterQueueBatcher",
-	featuregate.StageBeta,
+	featuregate.StageDeprecated,
 	featuregate.WithRegisterFromVersion("v0.115.0"),
-	featuregate.WithRegisterDescription("if set to true, turns on the pulling-based exporter queue bathcer"),
+	featuregate.WithRegisterToVersion("v0.119.0"),
+	featuregate.WithRegisterDescription("If set to true, turns on the pulling-based exporter queue bathcer"),
 )
 
 type ObsrepSenderFactory = func(obsrep *ObsReport) Sender[internal.Request]
@@ -101,23 +102,8 @@ func NewBaseExporter(set exporter.Settings, signal pipeline.Signal, osf ObsrepSe
 		be.QueueSender = NewQueueSender(q, be.Set, be.queueCfg.NumConsumers, be.ExportFailureMessage, be.Obsrep, be.BatcherCfg)
 	}
 
-	if !usePullingBasedExporterQueueBatcher.IsEnabled() && be.BatcherCfg.Enabled ||
-		usePullingBasedExporterQueueBatcher.IsEnabled() && be.BatcherCfg.Enabled && !be.queueCfg.Enabled {
-		bs := NewBatchSender(be.BatcherCfg, be.Set)
-		be.BatchSender = bs
-	}
-
 	be.connectSenders()
 
-	if bs, ok := be.BatchSender.(*BatchSender); ok {
-		// If queue sender is enabled assign to the batch sender the same number of workers.
-		if qs, ok := be.QueueSender.(*QueueSender); ok {
-			bs.concurrencyLimit = int64(qs.numConsumers)
-		}
-		// Batcher sender mutates the data.
-		be.ConsumerOptions = append(be.ConsumerOptions, consumer.WithCapabilities(consumer.Capabilities{MutatesData: true}))
-	}
-
 	return be, nil
 }
 

exporter/exporterhelper/internal/base_exporter_test.go

@@ -38,80 +38,52 @@ func newNoopObsrepSender(*ObsReport) Sender[internal.Request] {
 }
 
 func TestBaseExporter(t *testing.T) {
-	runTest := func(testName string, enableQueueBatcher bool) {
-		t.Run(testName, func(t *testing.T) {
-			defer setFeatureGateForTest(t, usePullingBasedExporterQueueBatcher, enableQueueBatcher)()
-			be, err := NewBaseExporter(defaultSettings, defaultSignal, newNoopObsrepSender)
-			require.NoError(t, err)
-			require.NoError(t, be.Start(context.Background(), componenttest.NewNopHost()))
-			require.NoError(t, be.Shutdown(context.Background()))
-		})
-	}
-	runTest("enable_queue_batcher", true)
-	runTest("disable_queue_batcher", false)
+	be, err := NewBaseExporter(defaultSettings, defaultSignal, newNoopObsrepSender)
+	require.NoError(t, err)
+	require.NoError(t, be.Start(context.Background(), componenttest.NewNopHost()))
+	require.NoError(t, be.Shutdown(context.Background()))
 }
 
 func TestBaseExporterWithOptions(t *testing.T) {
-	runTest := func(testName string, enableQueueBatcher bool) {
-		t.Run(testName, func(t *testing.T) {
-			defer setFeatureGateForTest(t, usePullingBasedExporterQueueBatcher, enableQueueBatcher)()
-			want := errors.New("my error")
-			be, err := NewBaseExporter(
-				defaultSettings, defaultSignal, newNoopObsrepSender,
-				WithStart(func(context.Context, component.Host) error { return want }),
-				WithShutdown(func(context.Context) error { return want }),
-				WithTimeout(NewDefaultTimeoutConfig()),
-			)
-			require.NoError(t, err)
-			require.Equal(t, want, be.Start(context.Background(), componenttest.NewNopHost()))
-			require.Equal(t, want, be.Shutdown(context.Background()))
-		})
-	}
-	runTest("enable_queue_batcher", true)
-	runTest("disable_queue_batcher", false)
+	want := errors.New("my error")
+	be, err := NewBaseExporter(
+		defaultSettings, defaultSignal, newNoopObsrepSender,
+		WithStart(func(context.Context, component.Host) error { return want }),
+		WithShutdown(func(context.Context) error { return want }),
+		WithTimeout(NewDefaultTimeoutConfig()),
+	)
+	require.NoError(t, err)
+	require.Equal(t, want, be.Start(context.Background(), componenttest.NewNopHost()))
+	require.Equal(t, want, be.Shutdown(context.Background()))
 }
 
 func TestQueueOptionsWithRequestExporter(t *testing.T) {
-	runTest := func(testName string, enableQueueBatcher bool) {
-		t.Run(testName, func(t *testing.T) {
-			defer setFeatureGateForTest(t, usePullingBasedExporterQueueBatcher, enableQueueBatcher)()
-			bs, err := NewBaseExporter(exportertest.NewNopSettings(), defaultSignal, newNoopObsrepSender,
-				WithRetry(configretry.NewDefaultBackOffConfig()))
-			require.NoError(t, err)
-			require.Nil(t, bs.Marshaler)
-			require.Nil(t, bs.Unmarshaler)
-			_, err = NewBaseExporter(exportertest.NewNopSettings(), defaultSignal, newNoopObsrepSender,
-				WithRetry(configretry.NewDefaultBackOffConfig()), WithQueue(NewDefaultQueueConfig()))
-			require.Error(t, err)
+	bs, err := NewBaseExporter(exportertest.NewNopSettings(), defaultSignal, newNoopObsrepSender,
+		WithRetry(configretry.NewDefaultBackOffConfig()))
+	require.NoError(t, err)
+	require.Nil(t, bs.Marshaler)
+	require.Nil(t, bs.Unmarshaler)
+	_, err = NewBaseExporter(exportertest.NewNopSettings(), defaultSignal, newNoopObsrepSender,
+		WithRetry(configretry.NewDefaultBackOffConfig()), WithQueue(NewDefaultQueueConfig()))
+	require.Error(t, err)
 
-			_, err = NewBaseExporter(exportertest.NewNopSettings(), defaultSignal, newNoopObsrepSender,
-				WithMarshaler(mockRequestMarshaler), WithUnmarshaler(mockRequestUnmarshaler(&mockRequest{})),
-				WithRetry(configretry.NewDefaultBackOffConfig()),
-				WithRequestQueue(exporterqueue.NewDefaultConfig(), exporterqueue.NewMemoryQueueFactory[internal.Request]()))
-			require.Error(t, err)
-		})
-	}
-	runTest("enable_queue_batcher", true)
-	runTest("disable_queue_batcher", false)
+	_, err = NewBaseExporter(exportertest.NewNopSettings(), defaultSignal, newNoopObsrepSender,
+		WithMarshaler(mockRequestMarshaler), WithUnmarshaler(mockRequestUnmarshaler(&mockRequest{})),
+		WithRetry(configretry.NewDefaultBackOffConfig()),
+		WithRequestQueue(exporterqueue.NewDefaultConfig(), exporterqueue.NewMemoryQueueFactory[internal.Request]()))
+	require.Error(t, err)
 }
 
 func TestBaseExporterLogging(t *testing.T) {
-	runTest := func(testName string, enableQueueBatcher bool) {
-		t.Run(testName, func(t *testing.T) {
-			defer setFeatureGateForTest(t, usePullingBasedExporterQueueBatcher, enableQueueBatcher)()
-			set := exportertest.NewNopSettings()
-			logger, observed := observer.New(zap.DebugLevel)
-			set.Logger = zap.New(logger)
-			rCfg := configretry.NewDefaultBackOffConfig()
-			rCfg.Enabled = false
-			bs, err := NewBaseExporter(set, defaultSignal, newNoopObsrepSender, WithRetry(rCfg))
-			require.NoError(t, err)
-			sendErr := bs.Send(context.Background(), newErrorRequest())
-			require.Error(t, sendErr)
+	set := exportertest.NewNopSettings()
+	logger, observed := observer.New(zap.DebugLevel)
+	set.Logger = zap.New(logger)
+	rCfg := configretry.NewDefaultBackOffConfig()
+	rCfg.Enabled = false
+	bs, err := NewBaseExporter(set, defaultSignal, newNoopObsrepSender, WithRetry(rCfg))
+	require.NoError(t, err)
+	sendErr := bs.Send(context.Background(), newErrorRequest())
+	require.Error(t, sendErr)
 
-			require.Len(t, observed.FilterLevelExact(zap.ErrorLevel).All(), 1)
-		})
-	}
-	runTest("enable_queue_batcher", true)
-	runTest("disable_queue_batcher", false)
+	require.Len(t, observed.FilterLevelExact(zap.ErrorLevel).All(), 1)
 }