[chore] Extract last export sender from timeout sender

exporter/exporterhelper/internal/base_exporter.go

@@ -70,8 +70,17 @@ func NewBaseExporter(set exporter.Settings, signal pipeline.Signal, options ...O
 		}
 	}
 
-	// TimeoutSender is always initialized.
-	be.firstSender = &TimeoutSender{cfg: be.timeoutCfg}
+	// Consumer Sender is always initialized.
+	be.firstSender = newSender(func(ctx context.Context, req request.Request) error {
+		return req.Export(ctx)
+	})
+
+	// Next setup the timeout Sender since we want the timeout to control only the export functionality.
+	// Only initialize if not explicitly disabled.
+	if be.timeoutCfg.Timeout != 0 {
+		be.firstSender = newTimeoutSender(be.timeoutCfg, be.firstSender)
+	}
+
 	if be.retryCfg.Enabled {
 		be.RetrySender = newRetrySender(be.retryCfg, set, be.firstSender)
 		be.firstSender = be.RetrySender

exporter/exporterhelper/internal/base_exporter_test.go

@@ -36,21 +36,15 @@ var (
 	}()
 )
 
-type noopSender struct {
-	component.StartFunc
-	component.ShutdownFunc
-	SendFunc[request.Request]
-}
-
 func newNoopExportSender() Sender[request.Request] {
-	return &noopSender{SendFunc: func(ctx context.Context, req request.Request) error {
+	return newSender(func(ctx context.Context, req request.Request) error {
 		select {
 		case <-ctx.Done():
 			return ctx.Err() // Returns the cancellation error
 		default:
 			return req.Export(ctx)
 		}
-	}}
+	})
 }
 
 func TestBaseExporter(t *testing.T) {

exporter/exporterhelper/internal/sender.go

@@ -0,0 +1,32 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package internal // import "go.opentelemetry.io/collector/exporter/exporterhelper/internal"
+
+import (
+	"context"
+
+	"go.opentelemetry.io/collector/component"
+)
+
+type Sender[K any] interface {
+	component.Component
+	Send(context.Context, K) error
+}
+
+type SendFunc[K any] func(ctx context.Context, data K) error
+
+// sender is a Sender that emits the incoming request to the exporter consumer func.
+type sender[K any] struct {
+	component.StartFunc
+	component.ShutdownFunc
+	consFunc SendFunc[K]
+}
+
+func newSender[K any](consFunc SendFunc[K]) Sender[K] {
+	return &sender[K]{consFunc: consFunc}
+}
+
+func (es *sender[K]) Send(ctx context.Context, req K) error {
+	return es.consFunc(ctx, req)
+}

exporter/exporterhelper/internal/sender_test.go

@@ -0,0 +1,30 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package internal
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestExportSenderRightArguments(t *testing.T) {
+	es := newSender[int64](func(_ context.Context, data int64) error {
+		assert.Equal(t, int64(1), data)
+		return nil
+	})
+	require.NoError(t, es.Send(context.Background(), int64(1)))
+}
+
+func TestExportSenderReturnsError(t *testing.T) {
+	err := errors.New("test error")
+	es := newSender[int64](func(_ context.Context, data int64) error {
+		assert.Equal(t, int64(1), data)
+		return err
+	})
+	require.ErrorIs(t, es.Send(context.Background(), int64(1)), err)
+}

exporter/exporterhelper/internal/timeout_sender.go

@@ -9,7 +9,6 @@ import (
 	"time"
 
 	"go.opentelemetry.io/collector/component"
-	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
 )
 
 // TimeoutConfig for timeout. The timeout applies to individual attempts to send data to the backend.
@@ -34,21 +33,22 @@ func NewDefaultTimeoutConfig() TimeoutConfig {
 	}
 }
 
-// TimeoutSender is a requestSender that adds a `timeout` to every request that passes this sender.
-type TimeoutSender struct {
+// timeoutSender is a requestSender that adds a `timeout` to every request that passes this sender.
+type timeoutSender[K any] struct {
 	component.StartFunc
 	component.ShutdownFunc
-	cfg TimeoutConfig
+	cfg  TimeoutConfig
+	next Sender[K]
 }
 
-func (ts *TimeoutSender) Send(ctx context.Context, req request.Request) error {
-	// TODO: Remove this by avoiding to create the timeout sender if timeout is 0.
-	if ts.cfg.Timeout == 0 {
-		return req.Export(ctx)
-	}
+func newTimeoutSender[K any](cfg TimeoutConfig, next Sender[K]) Sender[K] {
+	return &timeoutSender[K]{cfg: cfg, next: next}
+}
+
+func (ts *timeoutSender[K]) Send(ctx context.Context, req K) error {
 	// Intentionally don't overwrite the context inside the request, because in case of retries deadline will not be
 	// updated because this deadline most likely is before the next one.
 	tCtx, cancelFunc := context.WithTimeout(ctx, ts.cfg.Timeout)
 	defer cancelFunc()
-	return req.Export(tCtx)
+	return ts.next.Send(tCtx, req)
 }

exporter/exporterhelper/internal/timeout_sender_test.go

@@ -4,11 +4,14 @@
 package internal
 
 import (
+	"context"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+
+	"go.opentelemetry.io/collector/component/componenttest"
 )
 
 func TestNewDefaultTimeoutConfig(t *testing.T) {
@@ -23,3 +26,19 @@ func TestInvalidTimeout(t *testing.T) {
 	cfg.Timeout = -1
 	assert.Error(t, cfg.Validate())
 }
+
+func TestNewTimeoutSender(t *testing.T) {
+	cfg := TimeoutConfig{Timeout: 5 * time.Second}
+	ts := newTimeoutSender(cfg, newSender(func(ctx context.Context, data int64) error {
+		deadline, ok := ctx.Deadline()
+		assert.True(t, ok)
+		timeout := time.Since(deadline)
+		assert.LessOrEqual(t, timeout, 5*time.Second)
+		assert.GreaterOrEqual(t, 4*time.Second, timeout)
+		assert.Equal(t, int64(7), data)
+		return nil
+	}))
+	require.NoError(t, ts.Start(context.Background(), componenttest.NewNopHost()))
+	require.NoError(t, ts.Send(context.Background(), 7))
+	require.NoError(t, ts.Shutdown(context.Background()))
+}