Otel receivers spawn on need

CHANGELOG.md

@@ -34,6 +34,8 @@ Main (unreleased)
 
 - Fixed issue where text labels displayed outside of component node's boundary. (@hainenber)
 
+- Fixed a bug where a topic was claimed by the wrong consumer type in `otelcol.receiver.kafka`. (@wildum)
+
 ### Other changes
 
 - Update `alloy-mixin` to use more specific alert group names (for example,

docs/sources/reference/components/otelcol.receiver.kafka.md

@@ -279,6 +279,8 @@ Regular expressions are not allowed in the `headers` argument. Only exact matchi
 
 ### output block
 
+> **WARNING**: only one signal type should be set per `otelcol.receiver.kafka` component
+
 {{< docs/shared lookup="reference/components/output-block.md" source="alloy" version="<ALLOY_VERSION>" >}}
 
 ## Exported fields

internal/component/otelcol/receiver/kafka/kafka.go

@@ -2,8 +2,11 @@
 package kafka
 
 import (
+	"strings"
 	"time"
 
+	"github.com/go-kit/log"
+	"github.com/grafana/alloy/internal/alloy/logging/level"
 	"github.com/grafana/alloy/internal/component"
 	"github.com/grafana/alloy/internal/component/otelcol"
 	"github.com/grafana/alloy/internal/component/otelcol/receiver"
@@ -23,6 +26,7 @@ func init() {
 		Args:      Arguments{},
 
 		Build: func(opts component.Options, args component.Arguments) (component.Component, error) {
+			checkOutputSignals(args.(Arguments).Output, opts.Logger)
 			fact := kafkareceiver.NewFactory()
 			return receiver.New(opts, fact, args.(Arguments))
 		},
@@ -77,6 +81,23 @@ func (args *Arguments) SetToDefault() {
 	args.DebugMetrics.SetToDefault()
 }
 
+func checkOutputSignals(output *otelcol.ConsumerArguments, logger log.Logger) {
+	var signals []string
+
+	if len(output.Logs) > 0 {
+		signals = append(signals, "logs")
+	}
+	if len(output.Metrics) > 0 {
+		signals = append(signals, "metrics")
+	}
+	if len(output.Traces) > 0 {
+		signals = append(signals, "traces")
+	}
+	if len(signals) > 1 {
+		level.Warn(logger).Log("msg", "only one output signal should be set", "signals", strings.Join(signals, ", "))
+	}
+}
+
 // Convert implements receiver.Arguments.
 func (args Arguments) Convert() (otelcomponent.Config, error) {
 	input := make(map[string]interface{})

internal/component/otelcol/receiver/receiver.go

@@ -149,36 +149,40 @@ func (r *Receiver) Update(args component.Arguments) error {
 		return err
 	}
 
-	var (
-		next        = rargs.NextConsumers()
-		nextTraces  = fanoutconsumer.Traces(next.Traces)
-		nextMetrics = fanoutconsumer.Metrics(next.Metrics)
-		nextLogs    = fanoutconsumer.Logs(next.Logs)
-	)
+	next := rargs.NextConsumers()
 
 	// Create instances of the receiver from our factory for each of our
 	// supported telemetry signals.
 	var components []otelcomponent.Component
 
-	tracesReceiver, err := r.factory.CreateTracesReceiver(r.ctx, settings, receiverConfig, nextTraces)
-	if err != nil && !errors.Is(err, otelcomponent.ErrDataTypeIsNotSupported) {
-		return err
-	} else if tracesReceiver != nil {
-		components = append(components, tracesReceiver)
+	if len(next.Traces) > 0 {
+		nextTraces := fanoutconsumer.Traces(next.Traces)
+		tracesReceiver, err := r.factory.CreateTracesReceiver(r.ctx, settings, receiverConfig, nextTraces)
+		if err != nil && !errors.Is(err, otelcomponent.ErrDataTypeIsNotSupported) {
+			return err
+		} else if tracesReceiver != nil {
+			components = append(components, tracesReceiver)
+		}
 	}
 
-	metricsReceiver, err := r.factory.CreateMetricsReceiver(r.ctx, settings, receiverConfig, nextMetrics)
-	if err != nil && !errors.Is(err, otelcomponent.ErrDataTypeIsNotSupported) {
-		return err
-	} else if metricsReceiver != nil {
-		components = append(components, metricsReceiver)
+	if len(next.Metrics) > 0 {
+		nextMetrics := fanoutconsumer.Metrics(next.Metrics)
+		metricsReceiver, err := r.factory.CreateMetricsReceiver(r.ctx, settings, receiverConfig, nextMetrics)
+		if err != nil && !errors.Is(err, otelcomponent.ErrDataTypeIsNotSupported) {
+			return err
+		} else if metricsReceiver != nil {
+			components = append(components, metricsReceiver)
+		}
 	}
 
-	logsReceiver, err := r.factory.CreateLogsReceiver(r.ctx, settings, receiverConfig, nextLogs)
-	if err != nil && !errors.Is(err, otelcomponent.ErrDataTypeIsNotSupported) {
-		return err
-	} else if logsReceiver != nil {
-		components = append(components, logsReceiver)
+	if len(next.Logs) > 0 {
+		nextLogs := fanoutconsumer.Logs(next.Logs)
+		logsReceiver, err := r.factory.CreateLogsReceiver(r.ctx, settings, receiverConfig, nextLogs)
+		if err != nil && !errors.Is(err, otelcomponent.ErrDataTypeIsNotSupported) {
+			return err
+		} else if logsReceiver != nil {
+			components = append(components, logsReceiver)
+		}
 	}
 
 	// Schedule the components to run once our component is running.

internal/component/otelcol/receiver/receiver_test.go

@@ -57,6 +57,22 @@ func TestReceiver(t *testing.T) {
 	require.NoError(t, waitTracesTrigger.Wait(time.Second), "consumer did not get invoked")
 }
 
+func TestReceiverNotStarted(t *testing.T) {
+	var (
+		waitConsumerTrigger = util.NewWaitTrigger()
+		onTracesConsumer    = func(t otelconsumer.Traces) {
+			waitConsumerTrigger.Trigger()
+		}
+	)
+	te := newTestEnvironment(t, onTracesConsumer)
+	te.Start(fakeReceiverArgs{
+		Output: &otelcol.ConsumerArguments{},
+	})
+
+	// Check that no trace receiver was started because it's not needed by the output.
+	require.ErrorContains(t, waitConsumerTrigger.Wait(time.Second), "context deadline exceeded")
+}
+
 type testEnvironment struct {
 	t *testing.T