[receiver/kafkareceiver] fix: Kafka receiver blocking shutdown

.chloggen/fix-kafka-recv-blocking-shutdown.yaml

@@ -0,0 +1,27 @@
+# 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. filelogreceiver)
+component: kafkareceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Fixes issue causing kafkareceiver to block during Shutdown().
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [30789]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# 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: []

receiver/kafkareceiver/go.mod

@@ -17,7 +17,6 @@ require (
 	github.com/openzipkin/zipkin-go v0.4.3
 	github.com/stretchr/testify v1.9.0
 	go.opentelemetry.io/collector/component v0.111.1-0.20241008154146-ea48c09c31ae
-	go.opentelemetry.io/collector/component/componentstatus v0.111.1-0.20241008154146-ea48c09c31ae
 	go.opentelemetry.io/collector/config/configtelemetry v0.111.1-0.20241008154146-ea48c09c31ae
 	go.opentelemetry.io/collector/config/configtls v1.17.1-0.20241008154146-ea48c09c31ae
 	go.opentelemetry.io/collector/confmap v1.17.1-0.20241008154146-ea48c09c31ae

receiver/kafkareceiver/kafka_receiver.go

@@ -5,14 +5,12 @@ package kafkareceiver // import "github.com/open-telemetry/opentelemetry-collect
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"strconv"
 	"sync"
 
 	"github.com/IBM/sarama"
 	"go.opentelemetry.io/collector/component"
-	"go.opentelemetry.io/collector/component/componentstatus"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/pdata/pmetric"
@@ -207,16 +205,12 @@ func (c *kafkaTracesConsumer) Start(_ context.Context, host component.Host) erro
 			headers: c.headers,
 		}
 	}
-	go func() {
-		if err := c.consumeLoop(ctx, consumerGroup); !errors.Is(err, context.Canceled) {
-			componentstatus.ReportStatus(host, componentstatus.NewFatalErrorEvent(err))
-		}
-	}()
+	go c.consumeLoop(ctx, consumerGroup)
 	<-consumerGroup.ready
 	return nil
 }
 
-func (c *kafkaTracesConsumer) consumeLoop(ctx context.Context, handler sarama.ConsumerGroupHandler) error {
+func (c *kafkaTracesConsumer) consumeLoop(ctx context.Context, handler sarama.ConsumerGroupHandler) {
 	for {
 		// `Consume` should be called inside an infinite loop, when a
 		// server-side rebalance happens, the consumer session will need to be
@@ -227,7 +221,7 @@ func (c *kafkaTracesConsumer) consumeLoop(ctx context.Context, handler sarama.Co
 		// check if context was cancelled, signaling that the consumer should stop
 		if ctx.Err() != nil {
 			c.settings.Logger.Info("Consumer stopped", zap.Error(ctx.Err()))
-			return ctx.Err()
+			return
 		}
 	}
 }
@@ -315,16 +309,12 @@ func (c *kafkaMetricsConsumer) Start(_ context.Context, host component.Host) err
 			headers: c.headers,
 		}
 	}
-	go func() {
-		if err := c.consumeLoop(ctx, metricsConsumerGroup); err != nil {
-			componentstatus.ReportStatus(host, componentstatus.NewFatalErrorEvent(err))
-		}
-	}()
+	go c.consumeLoop(ctx, metricsConsumerGroup)
 	<-metricsConsumerGroup.ready
 	return nil
 }
 
-func (c *kafkaMetricsConsumer) consumeLoop(ctx context.Context, handler sarama.ConsumerGroupHandler) error {
+func (c *kafkaMetricsConsumer) consumeLoop(ctx context.Context, handler sarama.ConsumerGroupHandler) {
 	for {
 		// `Consume` should be called inside an infinite loop, when a
 		// server-side rebalance happens, the consumer session will need to be
@@ -335,7 +325,7 @@ func (c *kafkaMetricsConsumer) consumeLoop(ctx context.Context, handler sarama.C
 		// check if context was cancelled, signaling that the consumer should stop
 		if ctx.Err() != nil {
 			c.settings.Logger.Info("Consumer stopped", zap.Error(ctx.Err()))
-			return ctx.Err()
+			return
 		}
 	}
 }
@@ -426,16 +416,12 @@ func (c *kafkaLogsConsumer) Start(_ context.Context, host component.Host) error
 			headers: c.headers,
 		}
 	}
-	go func() {
-		if err := c.consumeLoop(ctx, logsConsumerGroup); err != nil {
-			componentstatus.ReportStatus(host, componentstatus.NewFatalErrorEvent(err))
-		}
-	}()
+	go c.consumeLoop(ctx, logsConsumerGroup)
 	<-logsConsumerGroup.ready
 	return nil
 }
 
-func (c *kafkaLogsConsumer) consumeLoop(ctx context.Context, handler sarama.ConsumerGroupHandler) error {
+func (c *kafkaLogsConsumer) consumeLoop(ctx context.Context, handler sarama.ConsumerGroupHandler) {
 	for {
 		// `Consume` should be called inside an infinite loop, when a
 		// server-side rebalance happens, the consumer session will need to be
@@ -446,7 +432,7 @@ func (c *kafkaLogsConsumer) consumeLoop(ctx context.Context, handler sarama.Cons
 		// check if context was cancelled, signaling that the consumer should stop
 		if ctx.Err() != nil {
 			c.settings.Logger.Info("Consumer stopped", zap.Error(ctx.Err()))
-			return ctx.Err()
+			return
 		}
 	}
 }

receiver/kafkareceiver/kafka_receiver_test.go

@@ -117,11 +117,10 @@ func TestTracesReceiverStartConsume(t *testing.T) {
 	ctx, cancelFunc := context.WithCancel(context.Background())
 	c.cancelConsumeLoop = cancelFunc
 	require.NoError(t, c.Shutdown(context.Background()))
-	err = c.consumeLoop(ctx, &tracesConsumerGroupHandler{
+	c.consumeLoop(ctx, &tracesConsumerGroupHandler{
 		ready:            make(chan bool),
 		telemetryBuilder: telemetryBuilder,
 	})
-	assert.EqualError(t, err, context.Canceled.Error())
 }
 
 func TestTracesReceiver_error(t *testing.T) {
@@ -456,11 +455,10 @@ func TestMetricsReceiverStartConsume(t *testing.T) {
 	ctx, cancelFunc := context.WithCancel(context.Background())
 	c.cancelConsumeLoop = cancelFunc
 	require.NoError(t, c.Shutdown(context.Background()))
-	err = c.consumeLoop(ctx, &logsConsumerGroupHandler{
+	c.consumeLoop(ctx, &logsConsumerGroupHandler{
 		ready:            make(chan bool),
 		telemetryBuilder: telemetryBuilder,
 	})
-	assert.EqualError(t, err, context.Canceled.Error())
 }
 
 func TestMetricsReceiver_error(t *testing.T) {
@@ -808,11 +806,10 @@ func TestLogsReceiverStartConsume(t *testing.T) {
 	ctx, cancelFunc := context.WithCancel(context.Background())
 	c.cancelConsumeLoop = cancelFunc
 	require.NoError(t, c.Shutdown(context.Background()))
-	err = c.consumeLoop(ctx, &logsConsumerGroupHandler{
+	c.consumeLoop(ctx, &logsConsumerGroupHandler{
 		ready:            make(chan bool),
 		telemetryBuilder: telemetryBuilder,
 	})
-	assert.EqualError(t, err, context.Canceled.Error())
 }
 
 func TestLogsReceiver_error(t *testing.T) {