[component] Deprecate ErrNilNextConsumer (#9526)

**Description:**
Deprecate ErrNilNextConsumer and implement nil checks in the builder
structs.

Author: atoulme

.chloggen/deprecate_ErrNilNextConsumer.yaml

@@ -0,0 +1,25 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: deprecation
+
+# The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver)
+component: component
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: deprecate component.ErrNilNextConsumer
+
+# One or more tracking issues or pull requests related to the change
+issues: [9526]
+
+# (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:
+
+# 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: [api]

component/component.go

@@ -11,6 +11,8 @@ import (
 var (
 	// ErrNilNextConsumer can be returned by receiver, or processor Start factory funcs that create the Component if the
 	// expected next Consumer is nil.
+	// Deprecated: [v0.96.0] The next consumer is now checked as part of the creation of the pipelines.
+	// This error will be removed in a future release.
 	ErrNilNextConsumer = errors.New("nil next Consumer")
 
 	// ErrDataTypeIsNotSupported can be returned by receiver, exporter or processor factory funcs that create the

connector/connector.go

@@ -5,6 +5,7 @@ package connector // import "go.opentelemetry.io/collector/connector"
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	"go.uber.org/zap"
@@ -13,6 +14,10 @@ import (
 	"go.opentelemetry.io/collector/consumer"
 )
 
+var (
+	errNilNextConsumer = errors.New("nil next Consumer")
+)
+
 // A Traces connector acts as an exporter from a traces pipeline and a receiver
 // to one or more traces, metrics, or logs pipelines.
 // Traces feeds a consumer.Traces, consumer.Metrics, or consumer.Logs with data.
@@ -456,6 +461,9 @@ func NewBuilder(cfgs map[component.ID]component.Config, factories map[component.
 
 // CreateTracesToTraces creates a Traces connector based on the settings and config.
 func (b *Builder) CreateTracesToTraces(ctx context.Context, set CreateSettings, next consumer.Traces) (Traces, error) {
+	if next == nil {
+		return nil, errNilNextConsumer
+	}
 	cfg, existsCfg := b.cfgs[set.ID]
 	if !existsCfg {
 		return nil, fmt.Errorf("connector %q is not configured", set.ID)
@@ -472,6 +480,9 @@ func (b *Builder) CreateTracesToTraces(ctx context.Context, set CreateSettings,
 
 // CreateTracesToMetrics creates a Traces connector based on the settings and config.
 func (b *Builder) CreateTracesToMetrics(ctx context.Context, set CreateSettings, next consumer.Metrics) (Traces, error) {
+	if next == nil {
+		return nil, errNilNextConsumer
+	}
 	cfg, existsCfg := b.cfgs[set.ID]
 	if !existsCfg {
 		return nil, fmt.Errorf("connector %q is not configured", set.ID)
@@ -488,6 +499,9 @@ func (b *Builder) CreateTracesToMetrics(ctx context.Context, set CreateSettings,
 
 // CreateTracesToLogs creates a Traces connector based on the settings and config.
 func (b *Builder) CreateTracesToLogs(ctx context.Context, set CreateSettings, next consumer.Logs) (Traces, error) {
+	if next == nil {
+		return nil, errNilNextConsumer
+	}
 	cfg, existsCfg := b.cfgs[set.ID]
 	if !existsCfg {
 		return nil, fmt.Errorf("connector %q is not configured", set.ID)
@@ -504,6 +518,9 @@ func (b *Builder) CreateTracesToLogs(ctx context.Context, set CreateSettings, ne
 
 // CreateMetricsToTraces creates a Metrics connector based on the settings and config.
 func (b *Builder) CreateMetricsToTraces(ctx context.Context, set CreateSettings, next consumer.Traces) (Metrics, error) {
+	if next == nil {
+		return nil, errNilNextConsumer
+	}
 	cfg, existsCfg := b.cfgs[set.ID]
 	if !existsCfg {
 		return nil, fmt.Errorf("connector %q is not configured", set.ID)
@@ -520,6 +537,9 @@ func (b *Builder) CreateMetricsToTraces(ctx context.Context, set CreateSettings,
 
 // CreateMetricsToMetrics creates a Metrics connector based on the settings and config.
 func (b *Builder) CreateMetricsToMetrics(ctx context.Context, set CreateSettings, next consumer.Metrics) (Metrics, error) {
+	if next == nil {
+		return nil, errNilNextConsumer
+	}
 	cfg, existsCfg := b.cfgs[set.ID]
 	if !existsCfg {
 		return nil, fmt.Errorf("connector %q is not configured", set.ID)
@@ -536,6 +556,9 @@ func (b *Builder) CreateMetricsToMetrics(ctx context.Context, set CreateSettings
 
 // CreateMetricsToLogs creates a Metrics connector based on the settings and config.
 func (b *Builder) CreateMetricsToLogs(ctx context.Context, set CreateSettings, next consumer.Logs) (Metrics, error) {
+	if next == nil {
+		return nil, errNilNextConsumer
+	}
 	cfg, existsCfg := b.cfgs[set.ID]
 	if !existsCfg {
 		return nil, fmt.Errorf("connector %q is not configured", set.ID)
@@ -552,6 +575,9 @@ func (b *Builder) CreateMetricsToLogs(ctx context.Context, set CreateSettings, n
 
 // CreateLogsToTraces creates a Logs connector based on the settings and config.
 func (b *Builder) CreateLogsToTraces(ctx context.Context, set CreateSettings, next consumer.Traces) (Logs, error) {
+	if next == nil {
+		return nil, errNilNextConsumer
+	}
 	cfg, existsCfg := b.cfgs[set.ID]
 	if !existsCfg {
 		return nil, fmt.Errorf("connector %q is not configured", set.ID)
@@ -568,6 +594,9 @@ func (b *Builder) CreateLogsToTraces(ctx context.Context, set CreateSettings, ne
 
 // CreateLogsToMetrics creates a Logs connector based on the settings and config.
 func (b *Builder) CreateLogsToMetrics(ctx context.Context, set CreateSettings, next consumer.Metrics) (Logs, error) {
+	if next == nil {
+		return nil, errNilNextConsumer
+	}
 	cfg, existsCfg := b.cfgs[set.ID]
 	if !existsCfg {
 		return nil, fmt.Errorf("connector %q is not configured", set.ID)
@@ -584,6 +613,9 @@ func (b *Builder) CreateLogsToMetrics(ctx context.Context, set CreateSettings, n
 
 // CreateLogsToLogs creates a Logs connector based on the settings and config.
 func (b *Builder) CreateLogsToLogs(ctx context.Context, set CreateSettings, next consumer.Logs) (Logs, error) {
+	if next == nil {
+		return nil, errNilNextConsumer
+	}
 	cfg, existsCfg := b.cfgs[set.ID]
 	if !existsCfg {
 		return nil, fmt.Errorf("connector %q is not configured", set.ID)