Fix start/stop logic for batch processor (#7708)

Signed-off-by: Bogdan Drutu <bogdandrutu@gmail.com>

.chloggen/fixbatcher.yaml

@@ -0,0 +1,11 @@
+# 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: batchprocessor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Fix start/stop logic for batch processor
+
+# One or more tracking issues or pull requests related to the change
+issues: [7708]

processor/batchprocessor/batch_processor.go

@@ -182,15 +182,11 @@ func (bp *batchProcessor) Capabilities() consumer.Capabilities {
 
 // Start is invoked during service startup.
 func (bp *batchProcessor) Start(context.Context, component.Host) error {
-	bp.goroutines.Add(1)
 	return nil
 }
 
 // Shutdown is invoked during service shutdown.
 func (bp *batchProcessor) Shutdown(context.Context) error {
-	// Done corresponds with the initial Add(1) in Start.
-	bp.goroutines.Done()
-
 	close(bp.shutdownC)
 
 	// Wait until all goroutines are done.

processor/batchprocessor/batch_processor_test.go

@@ -27,6 +27,57 @@ import (
 	"go.opentelemetry.io/collector/processor/processortest"
 )
 
+func TestProcessorShutdown(t *testing.T) {
+	factory := NewFactory()
+
+	ctx := context.Background()
+	processorCreationSet := processortest.NewNopCreateSettings()
+
+	for i := 0; i < 5; i++ {
+		require.NotPanics(t, func() {
+			tProc, err := factory.CreateTracesProcessor(ctx, processorCreationSet, factory.CreateDefaultConfig(), consumertest.NewNop())
+			require.NoError(t, err)
+			_ = tProc.Shutdown(ctx)
+		})
+
+		require.NotPanics(t, func() {
+			mProc, err := factory.CreateMetricsProcessor(ctx, processorCreationSet, factory.CreateDefaultConfig(), consumertest.NewNop())
+			require.NoError(t, err)
+			_ = mProc.Shutdown(ctx)
+		})
+
+		require.NotPanics(t, func() {
+			lProc, err := factory.CreateLogsProcessor(ctx, processorCreationSet, factory.CreateDefaultConfig(), consumertest.NewNop())
+			require.NoError(t, err)
+			_ = lProc.Shutdown(ctx)
+		})
+	}
+}
+
+func TestProcessorLifecycle(t *testing.T) {
+	factory := NewFactory()
+
+	ctx := context.Background()
+	processorCreationSet := processortest.NewNopCreateSettings()
+
+	for i := 0; i < 5; i++ {
+		tProc, err := factory.CreateTracesProcessor(ctx, processorCreationSet, factory.CreateDefaultConfig(), consumertest.NewNop())
+		require.NoError(t, err)
+		require.NoError(t, tProc.Start(ctx, componenttest.NewNopHost()))
+		require.NoError(t, tProc.Shutdown(ctx))
+
+		mProc, err := factory.CreateMetricsProcessor(ctx, processorCreationSet, factory.CreateDefaultConfig(), consumertest.NewNop())
+		require.NoError(t, err)
+		require.NoError(t, mProc.Start(ctx, componenttest.NewNopHost()))
+		require.NoError(t, mProc.Shutdown(ctx))
+
+		lProc, err := factory.CreateLogsProcessor(ctx, processorCreationSet, factory.CreateDefaultConfig(), consumertest.NewNop())
+		require.NoError(t, err)
+		require.NoError(t, lProc.Start(ctx, componenttest.NewNopHost()))
+		require.NoError(t, lProc.Shutdown(ctx))
+	}
+}
+
 func TestBatchProcessorSpansDelivered(t *testing.T) {
 	sink := new(consumertest.TracesSink)
 	cfg := createDefaultConfig().(*Config)