[exporterhelper] Fix batch sender ignoring next senders in the chain (#…

…10287)

This change fixes a bug when the retry and timeout logic was not applied
with enabled batching. The batch sender was ignoring the next senders in
the chain.

Fixes
#10166

.chloggen/fix_batch_sender_chaining.yaml

@@ -0,0 +1,20 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# 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: Fix a bug when the retry and timeout logic was not applied with enabled batching.
+
+# One or more tracking issues or pull requests related to the change
+issues: [10166]
+
+# 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: []

exporter/exporterhelper/batch_sender.go

@@ -119,7 +119,7 @@ func newEmptyBatch() *batch {
 // Caller must hold the lock.
 func (bs *batchSender) exportActiveBatch() {
 	go func(b *batch) {
-		b.err = b.request.Export(b.ctx)
+		b.err = bs.nextSender.send(b.ctx, b.request)
 		close(b.done)
 	}(bs.activeBatch)
 	bs.activeBatch = newEmptyBatch()
@@ -182,7 +182,7 @@ func (bs *batchSender) sendMergeSplitBatch(ctx context.Context, req Request) err
 	// Intentionally do not put the last request in the active batch to not block it.
 	// TODO: Consider including the partial request in the error to avoid double publishing.
 	for _, r := range reqs {
-		if err := r.Export(ctx); err != nil {
+		if err := bs.nextSender.send(ctx, r); err != nil {
 			return err
 		}
 	}

exporter/exporterhelper/batch_sender_test.go

@@ -483,6 +483,49 @@ func TestBatchSender_ShutdownDeadlock(t *testing.T) {
 	assert.EqualValues(t, 8, sink.itemsCount.Load())
 }
 
+func TestBatchSenderWithTimeout(t *testing.T) {
+	bCfg := exporterbatcher.NewDefaultConfig()
+	bCfg.MinSizeItems = 10
+	tCfg := NewDefaultTimeoutSettings()
+	tCfg.Timeout = 50 * time.Microsecond
+	be, err := newBaseExporter(defaultSettings, defaultDataType, newNoopObsrepSender,
+		WithBatcher(bCfg, WithRequestBatchFuncs(fakeBatchMergeFunc, fakeBatchMergeSplitFunc)),
+		WithTimeout(tCfg))
+	require.NoError(t, err)
+	require.NoError(t, be.Start(context.Background(), componenttest.NewNopHost()))
+
+	sink := newFakeRequestSink()
+
+	// Send 3 concurrent requests that should be merged in two batched
+	wg := sync.WaitGroup{}
+	for i := 0; i < 3; i++ {
+		wg.Add(1)
+		go func() {
+			require.NoError(t, be.send(context.Background(), &fakeRequest{items: 4, sink: sink}))
+			wg.Done()
+		}()
+	}
+	wg.Wait()
+	assert.EqualValues(t, 1, sink.requestsCount.Load())
+	assert.EqualValues(t, 12, sink.itemsCount.Load())
+
+	// 3 requests with a 90ms cumulative delay must be cancelled by the timeout sender
+	for i := 0; i < 3; i++ {
+		wg.Add(1)
+		go func() {
+			assert.Error(t, be.send(context.Background(), &fakeRequest{items: 4, sink: sink, delay: 30 * time.Millisecond}))
+			wg.Done()
+		}()
+	}
+	wg.Wait()
+
+	assert.NoError(t, be.Shutdown(context.Background()))
+
+	// The sink should not change
+	assert.EqualValues(t, 1, sink.requestsCount.Load())
+	assert.EqualValues(t, 12, sink.itemsCount.Load())
+}
+
 func queueBatchExporter(t *testing.T, batchOption Option) *baseExporter {
 	be, err := newBaseExporter(defaultSettings, defaultDataType, newNoopObsrepSender, batchOption,
 		WithRequestQueue(exporterqueue.NewDefaultConfig(), exporterqueue.NewMemoryQueueFactory[Request]()))