[batchprocessor] Batch processor reimplemented with Exporterhelper

processor/batchprocessor/capture_deadlock.sh

@@ -0,0 +1,36 @@
+#!/bin/bash
+
+# Script to capture deadlock stack trace from batch processor test
+
+echo "Starting deadlock capture..."
+cd /home/jmacd/src/otel/collector/processor/batchprocessor
+
+# Start the test in background and capture PID
+echo "Starting test in background..."
+go test -race -c .
+
+#TEST=TestBatchProcessorSpansDeliveredEnforceBatchSize/TestBatchProcessorSpansDeliveredEnforceBatchSizeHelper
+TEST=TestBatchProcessorSpansDeliveredEnforceBatchSize/TestBatchProcessorSpansDeliveredEnforceBatchSizeHelperWithPropagateErrors
+
+echo "Testing ${TEST}"
+./batchprocessor.test -test.v -test.run ${TEST} 2> deadlock_err 1> deadlock_out &
+TEST_PID=$!
+
+echo "Test started with PID: $TEST_PID"
+
+# Wait for the test to reach deadlock state
+echo "Waiting 10 seconds for deadlock to occur..."
+sleep 10
+
+# Send SIGABRT to get stack trace
+echo "Sending SIGABRT to capture stack trace..."
+kill -ABRT $TEST_PID
+
+# Wait a moment for the signal to be processed
+sleep 2
+
+kill -9 $TEST_PID 2>/dev/null
+
+cat deadlock_err
+echo "To access the whole stacktrace (also shown above), see ./deadlock_err"
+echo "To view test output see ./deadlock_out"

processor/batchprocessor/config.go

@@ -66,5 +66,11 @@ func (cfg *Config) Validate() error {
 	if cfg.Timeout < 0 {
 		return errors.New("timeout must be greater or equal to 0")
 	}
+
+	if useExporterHelper.IsEnabled() {
+		if len(cfg.MetadataKeys) != 0 {
+			return errors.New("metadata_keys not supported while this feature flag is alpha")
+		}
+	}
 	return nil
 }

processor/batchprocessor/exporterhelper_processor.go

@@ -0,0 +1,111 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package batchprocessor // import "go.opentelemetry.io/collector/processor/batchprocessor"
+
+import (
+	"context"
+	"runtime"
+
+	"go.opentelemetry.io/collector/config/configoptional"
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/exporter"
+	"go.opentelemetry.io/collector/exporter/exporterhelper"
+	"go.opentelemetry.io/collector/processor"
+)
+
+// translateToExporterHelperConfig converts legacy batchprocessor config to exporterhelper config
+func translateToExporterHelperConfig(cfg *Config) exporterhelper.QueueBatchConfig {
+	// These settings match legacy behavior
+	queueBatchConfig := exporterhelper.QueueBatchConfig{
+		Enabled:         true,
+		WaitForResult:   propagateErrors.IsEnabled(),
+		BlockOnOverflow: true,
+		Sizer:           exporterhelper.RequestSizerTypeItems,
+		QueueSize:       int64(runtime.NumCPU()) * int64(max(cfg.SendBatchSize, cfg.SendBatchMaxSize, 100)),
+		NumConsumers:    1,
+	}
+
+	if cfg.SendBatchSize > 0 || cfg.SendBatchMaxSize > 0 || cfg.Timeout > 0 {
+		batchConfig := exporterhelper.BatchConfig{
+			FlushTimeout: cfg.Timeout,
+			Sizer:        exporterhelper.RequestSizerTypeItems,
+			MinSize:      0, // Default: no minimum
+			MaxSize:      0, // Default: no maximum
+		}
+
+		// Map send_batch_size to MinSize (minimum items to trigger batch)
+		if cfg.SendBatchSize > 0 {
+			batchConfig.MinSize = int64(cfg.SendBatchSize)
+		}
+
+		// Map send_batch_max_size to MaxSize (maximum items in batch)
+		if cfg.SendBatchMaxSize > 0 {
+			batchConfig.MaxSize = int64(cfg.SendBatchMaxSize)
+		}
+
+		queueBatchConfig.Batch = configoptional.Some(batchConfig)
+	}
+
+	return queueBatchConfig
+}
+
+// newTracesProcessorWithExporterHelper creates a new traces processor using exporterhelper components.
+func newTracesProcessorWithExporterHelper(set processor.Settings, nextConsumer consumer.Traces, cfg *Config) (processor.Traces, error) {
+	set.Logger.Info("Creating traces processor with ExporterHelper")
+
+	queueBatchConfig := translateToExporterHelperConfig(cfg)
+
+	exporterSet := exporter.Settings{
+		ID:                set.ID,
+		TelemetrySettings: set.TelemetrySettings,
+		BuildInfo:         set.BuildInfo,
+	}
+
+	result, err := exporterhelper.NewTraces(
+		context.Background(),
+		exporterSet,
+		cfg,
+		nextConsumer.ConsumeTraces,
+		exporterhelper.WithQueue(queueBatchConfig),
+	)
+	return result, err
+}
+
+// newMetricsProcessorWithExporterHelper creates a new metrics processor using exporterhelper components.
+func newMetricsProcessorWithExporterHelper(set processor.Settings, nextConsumer consumer.Metrics, cfg *Config) (processor.Metrics, error) {
+	queueBatchConfig := translateToExporterHelperConfig(cfg)
+
+	exporterSet := exporter.Settings{
+		ID:                set.ID,
+		TelemetrySettings: set.TelemetrySettings,
+		BuildInfo:         set.BuildInfo,
+	}
+
+	return exporterhelper.NewMetrics(
+		context.Background(),
+		exporterSet,
+		cfg,
+		nextConsumer.ConsumeMetrics,
+		exporterhelper.WithQueue(queueBatchConfig),
+	)
+}
+
+// newLogsProcessorWithExporterHelper creates a new logs processor using exporterhelper components.
+func newLogsProcessorWithExporterHelper(set processor.Settings, nextConsumer consumer.Logs, cfg *Config) (processor.Logs, error) {
+	queueBatchConfig := translateToExporterHelperConfig(cfg)
+
+	exporterSet := exporter.Settings{
+		ID:                set.ID,
+		TelemetrySettings: set.TelemetrySettings,
+		BuildInfo:         set.BuildInfo,
+	}
+
+	return exporterhelper.NewLogs(
+		context.Background(),
+		exporterSet,
+		cfg,
+		nextConsumer.ConsumeLogs,
+		exporterhelper.WithQueue(queueBatchConfig),
+	)
+}

processor/batchprocessor/factory.go

@@ -49,7 +49,11 @@ func createTraces(
 	cfg component.Config,
 	nextConsumer consumer.Traces,
 ) (processor.Traces, error) {
-	return newTracesBatchProcessor(set, nextConsumer, cfg.(*Config))
+	batchCfg := cfg.(*Config)
+	if useExporterHelper.IsEnabled() {
+		return newTracesProcessorWithExporterHelper(set, nextConsumer, batchCfg)
+	}
+	return newTracesBatchProcessor(set, nextConsumer, batchCfg)
 }
 
 func createMetrics(
@@ -58,7 +62,11 @@ func createMetrics(
 	cfg component.Config,
 	nextConsumer consumer.Metrics,
 ) (processor.Metrics, error) {
-	return newMetricsBatchProcessor(set, nextConsumer, cfg.(*Config))
+	batchCfg := cfg.(*Config)
+	if useExporterHelper.IsEnabled() {
+		return newMetricsProcessorWithExporterHelper(set, nextConsumer, batchCfg)
+	}
+	return newMetricsBatchProcessor(set, nextConsumer, batchCfg)
 }
 
 func createLogs(
@@ -67,5 +75,9 @@ func createLogs(
 	cfg component.Config,
 	nextConsumer consumer.Logs,
 ) (processor.Logs, error) {
-	return newLogsBatchProcessor(set, nextConsumer, cfg.(*Config))
+	batchCfg := cfg.(*Config)
+	if useExporterHelper.IsEnabled() {
+		return newLogsProcessorWithExporterHelper(set, nextConsumer, batchCfg)
+	}
+	return newLogsBatchProcessor(set, nextConsumer, batchCfg)
 }

processor/batchprocessor/factory_test.go

@@ -10,6 +10,10 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/processor/processortest"
 )
 
@@ -22,21 +26,38 @@ func TestCreateDefaultConfig(t *testing.T) {
 }
 
 func TestCreateProcessor(t *testing.T) {
+	t.Run(t.Name() + "Legacy", func(t *testing.T) { testCreateProcessor(t, false) }) // Test legacy implementation
+	t.Run(t.Name() + "Helper", func(t *testing.T) { testCreateProcessor(t, true) })  // Test new implementation
+}
+
+func testCreateProcessor(t *testing.T, useExporterHelper bool) {
+	defer setFeatureGateForTest(t, useExporterHelper)()
+
 	factory := NewFactory()
 
 	cfg := factory.CreateDefaultConfig()
 	creationSet := processortest.NewNopSettings(factory.Type())
-	tp, err := factory.CreateTraces(context.Background(), creationSet, cfg, nil)
+
+	tc, _ := consumer.NewTraces(func(context.Context, ptrace.Traces) error {
+		return nil
+	})
+	tp, err := factory.CreateTraces(context.Background(), creationSet, cfg, tc)
 	assert.NotNil(t, tp)
 	assert.NoError(t, err, "cannot create trace processor")
 	assert.NoError(t, tp.Shutdown(context.Background()))
 
-	mp, err := factory.CreateMetrics(context.Background(), creationSet, cfg, nil)
+	mc, _ := consumer.NewMetrics(func(context.Context, pmetric.Metrics) error {
+		return nil
+	})
+	mp, err := factory.CreateMetrics(context.Background(), creationSet, cfg, mc)
 	assert.NotNil(t, mp)
 	assert.NoError(t, err, "cannot create metric processor")
 	assert.NoError(t, mp.Shutdown(context.Background()))
 
-	lp, err := factory.CreateLogs(context.Background(), creationSet, cfg, nil)
+	lc, _ := consumer.NewLogs(func(context.Context, plog.Logs) error {
+		return nil
+	})
+	lp, err := factory.CreateLogs(context.Background(), creationSet, cfg, lc)
 	assert.NotNil(t, lp)
 	assert.NoError(t, err, "cannot create logs processor")
 	assert.NoError(t, lp.Shutdown(context.Background()))

processor/batchprocessor/featuregates.go

@@ -0,0 +1,36 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package batchprocessor // import "go.opentelemetry.io/collector/processor/batchprocessor"
+
+import (
+	"go.opentelemetry.io/collector/featuregate"
+)
+
+const (
+	// useExporterHelperGate controls whether to use exporterhelper components
+	// for batching instead of the legacy implementation.
+	useExporterHelperGate = "processor.batch.useexporterhelper"
+
+	// propagateErrorsGate controls whether to propagate errors from the next
+	// consumer instead of suppressing them (legacy behavior).
+	propagateErrorsGate = "processor.batch.propagateerrors"
+)
+
+var (
+	// UseExporterHelper is the feature gate for using exporterhelper components.
+	useExporterHelper = featuregate.GlobalRegistry().MustRegister(
+		useExporterHelperGate,
+		featuregate.StageAlpha,
+		featuregate.WithRegisterDescription("Use exporterhelper components for batching instead of legacy implementation"),
+		featuregate.WithRegisterFromVersion("v0.131.0"),
+	)
+
+	// PropagateErrors is the feature gate for propagating errors instead of suppressing them.
+	propagateErrors = featuregate.GlobalRegistry().MustRegister(
+		propagateErrorsGate,
+		featuregate.StageAlpha,
+		featuregate.WithRegisterDescription("Propagate errors from next consumer instead of suppressing them"),
+		featuregate.WithRegisterFromVersion("v0.131.0"),
+	)
+)

processor/batchprocessor/go.mod

@@ -7,23 +7,28 @@ require (
 	go.opentelemetry.io/collector/client v1.37.0
 	go.opentelemetry.io/collector/component v1.37.0
 	go.opentelemetry.io/collector/component/componenttest v0.131.0
+	go.opentelemetry.io/collector/config/configoptional v0.131.0
 	go.opentelemetry.io/collector/confmap v1.37.0
 	go.opentelemetry.io/collector/consumer v1.37.0
 	go.opentelemetry.io/collector/consumer/consumererror v0.131.0
 	go.opentelemetry.io/collector/consumer/consumertest v0.131.0
+	go.opentelemetry.io/collector/exporter v1.36.1
+	go.opentelemetry.io/collector/featuregate v1.37.0
 	go.opentelemetry.io/collector/pdata v1.37.0
 	go.opentelemetry.io/collector/pdata/testdata v0.131.0
 	go.opentelemetry.io/collector/processor v1.37.0
-	go.opentelemetry.io/collector/processor/processortest v0.131.0
+	go.opentelemetry.io/collector/processor/processortest v0.130.1
 	go.opentelemetry.io/otel v1.37.0
 	go.opentelemetry.io/otel/metric v1.37.0
+	go.opentelemetry.io/otel/sdk v1.37.0
 	go.opentelemetry.io/otel/sdk/metric v1.37.0
 	go.opentelemetry.io/otel/trace v1.37.0
 	go.uber.org/goleak v1.3.0
 	go.uber.org/zap v1.27.0
 )
 
 require (
+	github.com/cenkalti/backoff/v5 v5.0.3 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/go-logr/logr v1.4.3 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
@@ -43,15 +48,17 @@ require (
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
 	go.opentelemetry.io/collector/component/componentstatus v0.131.0 // indirect
+	go.opentelemetry.io/collector/config/configretry v1.37.0 // indirect
 	go.opentelemetry.io/collector/consumer/xconsumer v0.131.0 // indirect
-	go.opentelemetry.io/collector/featuregate v1.37.0 // indirect
+	go.opentelemetry.io/collector/extension v1.37.0 // indirect
+	go.opentelemetry.io/collector/extension/xextension v0.131.0 // indirect
 	go.opentelemetry.io/collector/internal/telemetry v0.131.0 // indirect
 	go.opentelemetry.io/collector/pdata/pprofile v0.131.0 // indirect
+	go.opentelemetry.io/collector/pdata/xpdata v0.131.0 // indirect
 	go.opentelemetry.io/collector/pipeline v0.131.0 // indirect
 	go.opentelemetry.io/collector/processor/xprocessor v0.131.0 // indirect
 	go.opentelemetry.io/contrib/bridges/otelzap v0.12.0 // indirect
 	go.opentelemetry.io/otel/log v0.13.0 // indirect
-	go.opentelemetry.io/otel/sdk v1.37.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
 	golang.org/x/net v0.40.0 // indirect
@@ -103,3 +110,7 @@ replace go.opentelemetry.io/collector/consumer/consumererror => ../../consumer/c
 replace go.opentelemetry.io/collector/featuregate => ../../featuregate
 
 replace go.opentelemetry.io/collector/internal/telemetry => ../../internal/telemetry
+
+replace go.opentelemetry.io/collector/config/configretry => ../../config/configretry
+
+replace go.opentelemetry.io/collector/exporter => ../../exporter