kafkareceiver: connect async, report status (#40705)

#### Description

Make consumer group creation asynchronous (i.e. move it to background
goroutine), and report component status.

This change ensures that the collector can start up if the Kafka cluster
is not available or the consumer group creation fails due to some other
transient, retryable error. We also now report the component status to
indicate if and when the consumer group has been successfully created,
and whether the receiver is ready to consume messages. This can be
observed via the healthcheckv2 extension, e.g. for health/readiness
probes.

#### Link to tracking issue

Fixes #40516

#### Testing

First, I set up the collector with the following config:

```yaml
extensions:
  healthcheckv2:
    use_v2: true
    component_health:
      include_permanent_errors: true
      include_recoverable_errors: true
      recovery_duration: 30s
    http:
      endpoint: localhost:13133
      status:
        enabled: true
        path: /health/status
      config:
        enabled: true
        path: /health/config

receivers:
  kafka:
    brokers: [localhost:9092]

exporters:
  debug:            
    verbosity: detailed

service:
  extensions: [healthcheckv2]
  pipelines:
    metrics/kafka:
      receivers: [kafka]
      exporters: [debug]                 
```

Then:
 1. Without Kafka running locally, start the collector
2. Running `curl 'http://localhost:13133/health/status?verbose' | jq .`,
note that the receiver's status moves to StatusRecoverableError and
reports an error like "kafka: client has run out of available brokers to
talk to: dial tcp ..."
3. Start Kafka: `docker run --rm -p 9092:9092 --name broker
apache/kafka:4.0.0`
 4. Continue curling the status, and observe it moves to StatusOK
 5. Stop Kafka, observe status moves back to StatusRecoverableError

#### Documentation

No changes

---------

Co-authored-by: Sean Marciniak <[email protected]>

Author: axw

.chloggen/kafkareceiver-componentstatus.yaml

@@ -0,0 +1,33 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# 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: Create the consumer group asynchronously, and report component status
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [40516]
+
+# (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: |
+  This change ensures that the collector can start up if the Kafka cluster is not available
+  or the consumer group creation fails due to some other transient, retryable error.
+
+  We also now report the component status to indicate if and when the consumer group has been
+  successfully created, and whether the receiver is ready to consume messages. This can be
+  observed via the healthcheckv2 extension.
+
+# 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: [user]

receiver/kafkareceiver/consumer_sarama.go

@@ -13,6 +13,7 @@ import (
 	"github.com/IBM/sarama"
 	"github.com/cenkalti/backoff/v4"
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/component/componentstatus"
 	"go.opentelemetry.io/collector/consumer/consumererror"
 	"go.opentelemetry.io/collector/receiver"
 	"go.opentelemetry.io/collector/receiver/receiverhelper"
@@ -54,8 +55,8 @@ type saramaConsumer struct {
 	mu                sync.Mutex
 	started           bool
 	shutdown          bool
+	closing           chan struct{}
 	consumeLoopClosed chan struct{}
-	consumerGroup     sarama.ConsumerGroup
 }
 
 func (c *saramaConsumer) Start(_ context.Context, host component.Host) error {
@@ -77,20 +78,10 @@ func (c *saramaConsumer) Start(_ context.Context, host component.Host) error {
 		return err
 	}
 
-	consumerGroup, err := kafka.NewSaramaConsumerGroup(
-		context.Background(),
-		c.config.ClientConfig,
-		c.config.ConsumerConfig,
-	)
-	if err != nil {
-		return err
-	}
-	c.consumerGroup = consumerGroup
-
 	handler := &consumerGroupHandler{
+		host:              host,
 		id:                c.settings.ID,
 		logger:            c.settings.Logger,
-		ready:             make(chan bool),
 		obsrecv:           obsrecv,
 		autocommitEnabled: c.config.AutoCommit.Enable,
 		messageMarking:    c.config.MessageMarking,
@@ -105,42 +96,89 @@ func (c *saramaConsumer) Start(_ context.Context, host component.Host) error {
 
 	c.consumeLoopClosed = make(chan struct{})
 	c.started = true
-	go c.consumeLoop(handler)
+	c.closing = make(chan struct{})
+	go c.consumeLoop(handler, host)
 	return nil
 }
 
-func (c *saramaConsumer) consumeLoop(handler sarama.ConsumerGroupHandler) {
+func (c *saramaConsumer) consumeLoop(handler sarama.ConsumerGroupHandler, host component.Host) {
 	defer close(c.consumeLoopClosed)
+	defer componentstatus.ReportStatus(host, componentstatus.NewEvent(componentstatus.StatusStopped))
+	componentstatus.ReportStatus(host, componentstatus.NewEvent(componentstatus.StatusStarting))
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go func() {
+		select {
+		case <-ctx.Done():
+		case <-c.closing:
+			componentstatus.ReportStatus(host, componentstatus.NewEvent(componentstatus.StatusStopping))
+			cancel()
+		}
+	}()
+
+	// kafka.NewSaramaConsumerGroup (actually sarama.NewConsumerGroup)
+	// may perform synchronous operations that can fail due to transient
+	// errors, so we retry until it succeeds or the context is canceled.
+	var consumerGroup sarama.ConsumerGroup
+	err := backoff.Retry(func() (err error) {
+		consumerGroup, err = kafka.NewSaramaConsumerGroup(ctx, c.config.ClientConfig, c.config.ConsumerConfig)
+		if err != nil {
+			if ctx.Err() == nil {
+				// We only report an error if the context is not canceled.
+				// If the context is canceled it means the receiver is
+				// shutting down, which will lead to reporting StatusStopped
+				// when consumeLoop exits.
+				c.settings.Logger.Error("Error creating consumer group", zap.Error(err))
+				componentstatus.ReportStatus(host, componentstatus.NewRecoverableErrorEvent(err))
+			}
+			return err
+		}
+		return nil
+	}, backoff.WithContext(
+		// Use a zero max elapsed time to retry indefinitely until the context is canceled.
+		backoff.NewExponentialBackOff(backoff.WithMaxElapsedTime(0)),
+		ctx,
+	))
+	if err != nil {
+		return
+	}
+	defer func() {
+		if err := consumerGroup.Close(); err != nil {
+			c.settings.Logger.Error("Error closing consumer group", zap.Error(err))
+		}
+	}()
+	c.settings.Logger.Debug("Created consumer group")
 
-	ctx := context.Background()
 	for {
 		// `Consume` should be called inside an infinite loop, when a
 		// server-side rebalance happens, the consumer session will need to be
 		// recreated to get the new claims
-		if err := c.consumerGroup.Consume(ctx, c.topics, handler); err != nil {
+		if err := consumerGroup.Consume(ctx, c.topics, handler); err != nil {
+			if errors.Is(err, context.Canceled) {
+				// Shutting down
+				return
+			}
 			if errors.Is(err, sarama.ErrClosedConsumerGroup) {
-				c.settings.Logger.Info("Consumer stopped", zap.Error(ctx.Err()))
+				// Consumer group stopped unexpectedly.
+				c.settings.Logger.Warn("Consumer stopped", zap.Error(ctx.Err()))
 				return
 			}
 			c.settings.Logger.Error("Error from consumer", zap.Error(err))
+			componentstatus.ReportStatus(host, componentstatus.NewRecoverableErrorEvent(err))
 		}
 	}
 }
 
 func (c *saramaConsumer) Shutdown(ctx context.Context) error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
-	if c.shutdown {
+	if !c.started || c.shutdown {
 		return nil
 	}
 	c.shutdown = true
-	if !c.started {
-		return nil
-	}
+	close(c.closing)
 
-	if err := c.consumerGroup.Close(); err != nil {
-		return err
-	}
 	select {
 	case <-ctx.Done():
 		return ctx.Err()
@@ -150,10 +188,9 @@ func (c *saramaConsumer) Shutdown(ctx context.Context) error {
 }
 
 type consumerGroupHandler struct {
+	host           component.Host
 	id             component.ID
 	consumeMessage consumeMessageFunc
-	ready          chan bool
-	readyCloser    sync.Once
 	logger         *zap.Logger
 
 	obsrecv          *receiverhelper.ObsReport
@@ -166,22 +203,29 @@ type consumerGroupHandler struct {
 }
 
 func (c *consumerGroupHandler) Setup(session sarama.ConsumerGroupSession) error {
-	c.readyCloser.Do(func() { close(c.ready) })
+	c.logger.Debug("Consumer group session established")
+	componentstatus.ReportStatus(c.host, componentstatus.NewEvent(componentstatus.StatusOK))
 	c.telemetryBuilder.KafkaReceiverPartitionStart.Add(
 		session.Context(), 1, metric.WithAttributes(attribute.String(attrInstanceName, c.id.Name())),
 	)
 	return nil
 }
 
 func (c *consumerGroupHandler) Cleanup(session sarama.ConsumerGroupSession) error {
+	c.logger.Debug("Consumer group session stopped")
 	c.telemetryBuilder.KafkaReceiverPartitionClose.Add(
 		session.Context(), 1, metric.WithAttributes(attribute.String(attrInstanceName, c.id.Name())),
 	)
 	return nil
 }
 
 func (c *consumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
-	c.logger.Info("Starting consumer group", zap.Int32("partition", claim.Partition()))
+	c.logger.Debug(
+		"Consuming Kafka topic-partition",
+		zap.String("topic", claim.Topic()),
+		zap.Int32("partition", claim.Partition()),
+		zap.Int64("initial_offset", claim.InitialOffset()),
+	)
 	if !c.autocommitEnabled {
 		defer session.Commit()
 	}
@@ -242,8 +286,10 @@ func (c *consumerGroupHandler) handleMessage(
 					return err
 				}
 			}
-			c.logger.Warn("Stop error backoff because the configured max_elapsed_time is reached",
-				zap.Duration("max_elapsed_time", c.backOff.MaxElapsedTime))
+			c.logger.Warn(
+				"Stop error backoff because the configured max_elapsed_time is reached",
+				zap.Duration("max_elapsed_time", c.backOff.MaxElapsedTime),
+			)
 		}
 		if c.messageMarking.After && !c.messageMarking.OnError {
 			// Only return an error if messages are marked after successful processing.

receiver/kafkareceiver/factory_test.go

@@ -30,8 +30,8 @@ func TestCreateTraces(t *testing.T) {
 	cfg.ProtocolVersion = "2.0.0"
 	r, err := createTracesReceiver(context.Background(), receivertest.NewNopSettings(metadata.Type), cfg, nil)
 	require.NoError(t, err)
-	// no available broker
-	require.Error(t, r.Start(context.Background(), componenttest.NewNopHost()))
+	require.NoError(t, r.Start(context.Background(), componenttest.NewNopHost()))
+	assert.NoError(t, r.Shutdown(context.Background()))
 }
 
 func TestWithTracesUnmarshalers(t *testing.T) {
@@ -64,8 +64,8 @@ func TestCreateMetrics(t *testing.T) {
 	cfg.ProtocolVersion = "2.0.0"
 	r, err := createMetricsReceiver(context.Background(), receivertest.NewNopSettings(metadata.Type), cfg, nil)
 	require.NoError(t, err)
-	// no available broker
-	require.Error(t, r.Start(context.Background(), componenttest.NewNopHost()))
+	require.NoError(t, r.Start(context.Background(), componenttest.NewNopHost()))
+	assert.NoError(t, r.Shutdown(context.Background()))
 }
 
 func TestWithMetricsUnmarshalers(t *testing.T) {
@@ -98,8 +98,8 @@ func TestCreateLogs(t *testing.T) {
 	cfg.ProtocolVersion = "2.0.0"
 	r, err := createLogsReceiver(context.Background(), receivertest.NewNopSettings(metadata.Type), cfg, nil)
 	require.NoError(t, err)
-	// no available broker
-	require.Error(t, r.Start(context.Background(), componenttest.NewNopHost()))
+	require.NoError(t, r.Start(context.Background(), componenttest.NewNopHost()))
+	assert.NoError(t, r.Shutdown(context.Background()))
 }
 
 func TestWithLogsUnmarshalers(t *testing.T) {

receiver/kafkareceiver/go.mod

@@ -22,6 +22,7 @@ require (
 	github.com/twmb/franz-go/pkg/kfake v0.0.0-20250320172111-35ab5e5f5327
 	go.opentelemetry.io/collector/client v1.34.1-0.20250610090210-188191247685
 	go.opentelemetry.io/collector/component v1.34.1-0.20250610090210-188191247685
+	go.opentelemetry.io/collector/component/componentstatus v0.128.1-0.20250610090210-188191247685
 	go.opentelemetry.io/collector/component/componenttest v0.128.1-0.20250610090210-188191247685
 	go.opentelemetry.io/collector/config/configretry v1.34.1-0.20250610090210-188191247685
 	go.opentelemetry.io/collector/config/configtls v1.34.1-0.20250610090210-188191247685