Stabilize exporter.UsePullingBasedExporterQueueBatcher and remove old batch sender

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

Author: bogdandrutu

.chloggen/rm-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: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver)
+component: exporterhelper
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Stabilize exporter.UsePullingBasedExporterQueueBatcher and remove old batch sender
+
+# One or more tracking issues or pull requests related to the change
+issues: []
+
+# (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: [api, user]

exporter/exporterhelper/internal/base_exporter_test.go

@@ -48,94 +48,66 @@ func newNoopExportSender() Sender[request.Request] {
 }
 
 func TestBaseExporter(t *testing.T) {
-	runTest := func(testName string, enableQueueBatcher bool) {
-		t.Run(testName, func(t *testing.T) {
-			setFeatureGateForTest(t, usePullingBasedExporterQueueBatcher, enableQueueBatcher)
-			be, err := NewBaseExporter(defaultSettings, defaultSignal)
-			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)
+	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) {
-			setFeatureGateForTest(t, usePullingBasedExporterQueueBatcher, enableQueueBatcher)
-			want := errors.New("my error")
-			be, err := NewBaseExporter(
-				defaultSettings, defaultSignal,
-				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,
+		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) {
-			setFeatureGateForTest(t, usePullingBasedExporterQueueBatcher, enableQueueBatcher)
-			bs, err := NewBaseExporter(exportertest.NewNopSettingsWithType(exportertest.NopType), defaultSignal,
-				WithRetry(configretry.NewDefaultBackOffConfig()))
-			require.NoError(t, err)
-			require.Nil(t, bs.Marshaler)
-			require.Nil(t, bs.Unmarshaler)
-			_, err = NewBaseExporter(exportertest.NewNopSettingsWithType(exportertest.NopType), defaultSignal,
-				WithRetry(configretry.NewDefaultBackOffConfig()), WithQueue(NewDefaultQueueConfig()))
-			require.Error(t, err)
+	bs, err := NewBaseExporter(exportertest.NewNopSettingsWithType(exportertest.NopType), defaultSignal,
+		WithRetry(configretry.NewDefaultBackOffConfig()))
+	require.NoError(t, err)
+	require.Nil(t, bs.Marshaler)
+	require.Nil(t, bs.Unmarshaler)
+	_, err = NewBaseExporter(exportertest.NewNopSettingsWithType(exportertest.NopType), defaultSignal,
+		WithRetry(configretry.NewDefaultBackOffConfig()), WithQueue(NewDefaultQueueConfig()))
+	require.Error(t, err)
 
-			_, err = NewBaseExporter(exportertest.NewNopSettingsWithType(exportertest.NopType), defaultSignal,
-				WithMarshaler(mockRequestMarshaler), WithUnmarshaler(mockRequestUnmarshaler(&requesttest.FakeRequest{Items: 1})),
-				WithRetry(configretry.NewDefaultBackOffConfig()),
-				WithRequestQueue(exporterqueue.NewDefaultConfig(), exporterqueue.NewMemoryQueueFactory[request.Request]()))
-			require.Error(t, err)
-		})
-	}
-	runTest("enable_queue_batcher", true)
-	runTest("disable_queue_batcher", false)
+	_, err = NewBaseExporter(exportertest.NewNopSettingsWithType(exportertest.NopType), defaultSignal,
+		WithMarshaler(mockRequestMarshaler), WithUnmarshaler(mockRequestUnmarshaler(&requesttest.FakeRequest{Items: 1})),
+		WithRetry(configretry.NewDefaultBackOffConfig()),
+		WithRequestQueue(exporterqueue.NewDefaultConfig(), exporterqueue.NewMemoryQueueFactory[request.Request]()))
+	require.Error(t, err)
 }
 
 func TestBaseExporterLogging(t *testing.T) {
-	runTest := func(testName string, enableQueueBatcher bool) {
-		t.Run(testName, func(t *testing.T) {
-			setFeatureGateForTest(t, usePullingBasedExporterQueueBatcher, enableQueueBatcher)
-			set := exportertest.NewNopSettingsWithType(exportertest.NopType)
-			logger, observed := observer.New(zap.DebugLevel)
-			set.Logger = zap.New(logger)
-			rCfg := configretry.NewDefaultBackOffConfig()
-			rCfg.Enabled = false
-			qCfg := exporterqueue.NewDefaultConfig()
-			qCfg.Enabled = false
-			bs, err := NewBaseExporter(set, defaultSignal,
-				WithRequestQueue(qCfg, exporterqueue.NewMemoryQueueFactory[request.Request]()),
-				WithBatcher(exporterbatcher.NewDefaultConfig()),
-				WithRetry(rCfg))
-			require.NoError(t, err)
-			require.NoError(t, bs.Start(context.Background(), componenttest.NewNopHost()))
-			sink := requesttest.NewSink()
-			sendErr := bs.Send(context.Background(), &requesttest.FakeRequest{Items: 2, Sink: sink, ExportErr: errors.New("my error")})
-			require.Error(t, sendErr)
+	set := exportertest.NewNopSettingsWithType(exportertest.NopType)
+	logger, observed := observer.New(zap.DebugLevel)
+	set.Logger = zap.New(logger)
+	rCfg := configretry.NewDefaultBackOffConfig()
+	rCfg.Enabled = false
+	qCfg := exporterqueue.NewDefaultConfig()
+	qCfg.Enabled = false
+	bs, err := NewBaseExporter(set, defaultSignal,
+		WithRequestQueue(qCfg, exporterqueue.NewMemoryQueueFactory[request.Request]()),
+		WithBatcher(exporterbatcher.NewDefaultConfig()),
+		WithRetry(rCfg))
+	require.NoError(t, err)
+	require.NoError(t, bs.Start(context.Background(), componenttest.NewNopHost()))
+	sink := requesttest.NewSink()
+	sendErr := bs.Send(context.Background(), &requesttest.FakeRequest{Items: 2, Sink: sink, ExportErr: errors.New("my error")})
+	require.Error(t, sendErr)
 
-			require.Len(t, observed.FilterLevelExact(zap.ErrorLevel).All(), 2)
-			assert.Contains(t, observed.All()[0].Message, "Exporting failed. Dropping data.")
-			assert.Equal(t, "my error", observed.All()[0].ContextMap()["error"])
-			assert.Contains(t, observed.All()[1].Message, "Exporting failed. Rejecting data.")
-			assert.Equal(t, "my error", observed.All()[1].ContextMap()["error"])
-			require.NoError(t, bs.Shutdown(context.Background()))
-		})
-	}
-	runTest("enable_queue_batcher", true)
-	runTest("disable_queue_batcher", false)
+	require.Len(t, observed.FilterLevelExact(zap.ErrorLevel).All(), 2)
+	assert.Contains(t, observed.All()[0].Message, "Exporting failed. Dropping data.")
+	assert.Equal(t, "my error", observed.All()[0].ContextMap()["error"])
+	assert.Contains(t, observed.All()[1].Message, "Exporting failed. Rejecting data.")
+	assert.Equal(t, "my error", observed.All()[1].ContextMap()["error"])
+	require.NoError(t, bs.Shutdown(context.Background()))
 }
 
 func TestQueueRetryWithDisabledQueue(t *testing.T) {
@@ -177,13 +149,8 @@ func TestQueueRetryWithDisabledQueue(t *testing.T) {
 		},
 	}
 
-	runTest := func(testName string, enableQueueBatcher bool, tt struct {
-		name         string
-		queueOptions []Option
-	},
-	) {
-		t.Run(testName, func(t *testing.T) {
-			setFeatureGateForTest(t, usePullingBasedExporterQueueBatcher, enableQueueBatcher)
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
 			set := exportertest.NewNopSettingsWithType(exportertest.NopType)
 			logger, observed := observer.New(zap.ErrorLevel)
 			set.Logger = zap.New(logger)
@@ -199,8 +166,4 @@ func TestQueueRetryWithDisabledQueue(t *testing.T) {
 			assert.Empty(t, 0, sink.RequestsCount())
 		})
 	}
-	for _, tt := range tests {
-		runTest(tt.name+"_enable_queue_batcher", true, tt)
-		runTest(tt.name+"_disable_queue_batcher", false, tt)
-	}
 }