Incorporate review comments

Author: raghu999

exporter/exporterhelper/internal/queue_sender.go

@@ -21,6 +21,12 @@ import (
 )
 
 // NewDefaultQueueConfig returns the default config for queuebatch.Config.
+// By default:
+//
+// - the queue stores 1000 requests of telemetry
+// - is non-blocking when full
+// - concurrent exports limited to 10
+// - emits batches of 8192 items, timeout 200ms
 func NewDefaultQueueConfig() queuebatch.Config {
 	return queuebatch.Config{
 		Sizer:           request.SizerTypeRequests,
@@ -89,36 +95,9 @@ func NewQueueSender(
 	}
 	qs.QueueBatch = qb
 
-	if len(qCfg.RequestMiddlewares) > 0 {
-		warnIfNumConsumersMayCapMiddleware(&qCfg, qSet.Telemetry.Logger)
-	}
-
 	return qs, nil
 }
 
-// warnIfNumConsumersMayCapMiddleware ensures sufficient worker headroom when request middleware is configured.
-func warnIfNumConsumersMayCapMiddleware(cfg *queuebatch.Config, logger *zap.Logger) {
-	if len(cfg.RequestMiddlewares) == 0 {
-		return
-	}
-
-	defaultConsumers := NewDefaultQueueConfig().NumConsumers
-
-	// If the user left the default worker count, warn that the worker pool might artificially cap the controller.
-	if cfg.NumConsumers == defaultConsumers && logger != nil {
-		logger.Warn("sending_queue.num_consumers is at the default; request middleware may be capped by worker pool",
-			zap.Int("num_consumers", cfg.NumConsumers),
-			zap.Strings("request_middlewares", func() []string {
-				ids := make([]string, len(cfg.RequestMiddlewares))
-				for i, id := range cfg.RequestMiddlewares {
-					ids[i] = id.String()
-				}
-				return ids
-			}()),
-		)
-	}
-}
-
 // Start overrides the default Start to resolve the extensions and wrap the sender.
 func (qs *queueSender) Start(ctx context.Context, host component.Host) error {
 	mws := make([]requestmiddleware.RequestMiddleware, 0, len(qs.mwIDs))

exporter/exporterhelper/internal/queue_sender_test.go

@@ -62,8 +62,6 @@ func TestQueueConfig_Validate(t *testing.T) {
 	assert.NoError(t, noCfg.Validate())
 }
 
-// --- New Tests for Request Middleware ---
-
 // nopComponent satisfies component.Component but does not implement RequestMiddleware.
 type nopComponent struct{}
 
@@ -299,100 +297,3 @@ func TestQueueSender_Start_Errors_And_Cleanup(t *testing.T) {
 		})
 	}
 }
-
-func TestNewQueueSender_RequestMiddleware_LogsWarningOnDefaultConsumers(t *testing.T) {
-	mwID := component.MustNewID("request_middleware")
-
-	tests := []struct {
-		name          string
-		inputConfig   queuebatch.Config
-		wantConsumers int
-		wantLog       bool
-	}{
-		{
-			name: "warn_on_default_consumer_count",
-			inputConfig: func() queuebatch.Config {
-				cfg := NewDefaultQueueConfig()
-				cfg.RequestMiddlewares = []component.ID{mwID}
-				cfg.NumConsumers = 10
-				return cfg
-			}(),
-			wantConsumers: 10, // Should NOT change
-			wantLog:       true,
-		},
-		{
-			name: "respect_high_consumer_count",
-			inputConfig: func() queuebatch.Config {
-				cfg := NewDefaultQueueConfig()
-				cfg.RequestMiddlewares = []component.ID{mwID}
-				cfg.NumConsumers = 500
-				return cfg
-			}(),
-			wantConsumers: 500,
-			wantLog:       false,
-		},
-		{
-			name: "ignore_if_middleware_disabled",
-			inputConfig: func() queuebatch.Config {
-				cfg := NewDefaultQueueConfig()
-				cfg.RequestMiddlewares = nil
-				cfg.NumConsumers = 10
-				return cfg
-			}(),
-			wantConsumers: 10,
-			wantLog:       false,
-		},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			logger, observed := observer.New(zap.WarnLevel)
-			cfg := tt.inputConfig
-
-			warnIfNumConsumersMayCapMiddleware(&cfg, zap.New(logger))
-
-			assert.Equal(t, tt.wantConsumers, cfg.NumConsumers)
-			if tt.wantLog {
-				require.Len(t, observed.All(), 1, "Expected a warning log")
-				assert.Contains(t, observed.All()[0].Message, "sending_queue.num_consumers is at the default")
-			} else {
-				assert.Len(t, observed.All(), 0, "Expected no warning logs")
-			}
-		})
-	}
-}
-
-func TestQueueSender_RequestMiddleware_Disabled_NoInteraction(t *testing.T) {
-	// 1. Setup a host with the extension available, but do NOT use it.
-	mwID := component.MustNewID("request_middleware")
-	mockMw := &mockRequestMiddleware{}
-	host := &mockHost{
-		ext: map[component.ID]component.Component{
-			mwID: mockMw,
-		},
-	}
-
-	// 2. do NOT configure the RequestMiddlewares in the queue settings.
-	qSet := queuebatch.AllSettings[request.Request]{
-		ID:        component.MustNewID("otlp"),
-		Signal:    pipeline.SignalTraces,
-		Telemetry: componenttest.NewNopTelemetrySettings(),
-	}
-	qCfg := NewDefaultQueueConfig()
-	qCfg.RequestMiddlewares = nil
-
-	nextSender := sender.NewSender(func(ctx context.Context, req request.Request) error {
-		return nil
-	})
-
-	qs, err := NewQueueSender(qSet, qCfg, "", nextSender)
-	require.NoError(t, err)
-
-	// 3. Start and Send data
-	require.NoError(t, qs.Start(context.Background(), host))
-	require.NoError(t, qs.Send(context.Background(), &requesttest.FakeRequest{Items: 10}))
-	require.NoError(t, qs.Shutdown(context.Background()))
-
-	// 4. Verification: Assert ABSOLUTELY NO interaction with the mock middleware
-	assert.Nil(t, mockMw.wrapper, "WrapSender should not be called when middleware is disabled")
-}

exporter/exporterhelper/internal/requestmiddleware/interfaces.go

@@ -0,0 +1,28 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package requestmiddleware // import "go.opentelemetry.io/collector/exporter/exporterhelper/internal/requestmiddleware"
+
+import (
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/internal/sender"
+	"go.opentelemetry.io/collector/pipeline"
+)
+
+// RequestMiddleware enables extensions to intercept or modify exporter requests.
+// Implementations can wrap the downstream sender with additional logic (e.g., rate control, adaptive concurrency).
+
+// RequestMiddlewareSettings provides context about the exporter request pipeline.
+type RequestMiddlewareSettings struct {
+	Signal    pipeline.Signal
+	ID        component.ID
+	Telemetry component.TelemetrySettings
+}
+
+// RequestMiddleware allows an extension to wrap an exporter's request sender.
+type RequestMiddleware interface {
+	// WrapSender allows the middleware to wrap the next sender in the chain.
+	// It returns a generic sender that implements the specific request type.
+	WrapSender(set RequestMiddlewareSettings, next sender.Sender[request.Request]) (sender.Sender[request.Request], error)
+}