Remove checks of ErrNilNextConsumer

.chloggen/remove_ErrNilNextConsumer_impl.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: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: all
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Remove explicit checks in all receivers to check if the next consumer is nil
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [31793]
+
+# (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: The nil check is now done by the pipeline builder.
+
+# 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: [api]

processor/tailsamplingprocessor/processor.go

@@ -77,10 +77,6 @@ const (
 // newTracesProcessor returns a processor.TracesProcessor that will perform tail sampling according to the given
 // configuration.
 func newTracesProcessor(ctx context.Context, settings component.TelemetrySettings, nextConsumer consumer.Traces, cfg Config) (processor.Traces, error) {
-	if nextConsumer == nil {
-		return nil, component.ErrNilNextConsumer
-	}
-
 	policyNames := map[string]bool{}
 	policies := make([]*policy, len(cfg.PolicyCfgs))
 	for i := range cfg.PolicyCfgs {

receiver/awscontainerinsightreceiver/receiver.go

@@ -44,9 +44,6 @@ func newAWSContainerInsightReceiver(
 	settings component.TelemetrySettings,
 	config *Config,
 	nextConsumer consumer.Metrics) (receiver.Metrics, error) {
-	if nextConsumer == nil {
-		return nil, component.ErrNilNextConsumer
-	}
 
 	r := &awsContainerInsightReceiver{
 		settings:     settings,

receiver/awscontainerinsightreceiver/receiver_test.go

@@ -63,18 +63,6 @@ func TestReceiver(t *testing.T) {
 	require.NoError(t, err)
 }
 
-func TestReceiverForNilConsumer(t *testing.T) {
-	cfg := createDefaultConfig().(*Config)
-	metricsReceiver, err := newAWSContainerInsightReceiver(
-		componenttest.NewNopTelemetrySettings(),
-		cfg,
-		nil,
-	)
-
-	require.Error(t, err)
-	require.Nil(t, metricsReceiver)
-}
-
 func TestCollectData(t *testing.T) {
 	cfg := createDefaultConfig().(*Config)
 	metricsReceiver, err := newAWSContainerInsightReceiver(

receiver/awsecscontainermetricsreceiver/receiver.go

@@ -34,10 +34,6 @@ func newAWSECSContainermetrics(
 	config *Config,
 	nextConsumer consumer.Metrics,
 	rest ecsutil.RestClient) (receiver.Metrics, error) {
-	if nextConsumer == nil {
-		return nil, component.ErrNilNextConsumer
-	}
-
 	r := &awsEcsContainerMetricsReceiver{
 		logger:       logger,
 		nextConsumer: nextConsumer,

receiver/awsecscontainermetricsreceiver/receiver_test.go

@@ -53,19 +53,6 @@ func TestReceiver(t *testing.T) {
 	require.NoError(t, err)
 }
 
-func TestReceiverForNilConsumer(t *testing.T) {
-	cfg := createDefaultConfig().(*Config)
-	metricsReceiver, err := newAWSECSContainermetrics(
-		zap.NewNop(),
-		cfg,
-		nil,
-		&fakeRestClient{},
-	)
-
-	require.Error(t, err)
-	require.Nil(t, metricsReceiver)
-}
-
 func TestCollectDataFromEndpoint(t *testing.T) {
 	cfg := createDefaultConfig().(*Config)
 	metricsReceiver, err := newAWSECSContainermetrics(

receiver/awsfirehosereceiver/metrics_receiver.go

@@ -7,7 +7,6 @@ import (
 	"context"
 	"net/http"
 
-	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/receiver"
 
@@ -35,9 +34,6 @@ func newMetricsReceiver(
 	unmarshalers map[string]unmarshaler.MetricsUnmarshaler,
 	nextConsumer consumer.Metrics,
 ) (receiver.Metrics, error) {
-	if nextConsumer == nil {
-		return nil, component.ErrNilNextConsumer
-	}
 
 	configuredUnmarshaler := unmarshalers[config.RecordType]
 	if configuredUnmarshaler == nil {

receiver/awsxrayreceiver/receiver.go

@@ -42,10 +42,6 @@ func newReceiver(config *Config,
 	consumer consumer.Traces,
 	set receiver.CreateSettings) (receiver.Traces, error) {
 
-	if consumer == nil {
-		return nil, component.ErrNilNextConsumer
-	}
-
 	set.Logger.Info("Going to listen on endpoint for X-Ray segments",
 		zap.String(udppoller.Transport, config.Endpoint))
 	poller, err := udppoller.New(&udppoller.Config{

receiver/carbonreceiver/receiver.go

@@ -42,10 +42,6 @@ func newMetricsReceiver(
 	nextConsumer consumer.Metrics,
 ) (receiver.Metrics, error) {
 
-	if nextConsumer == nil {
-		return nil, component.ErrNilNextConsumer
-	}
-
 	if config.Endpoint == "" {
 		return nil, errEmptyEndpoint
 	}

receiver/cloudfoundryreceiver/receiver.go

@@ -43,10 +43,6 @@ func newCloudFoundryReceiver(
 	config Config,
 	nextConsumer consumer.Metrics) (receiver.Metrics, error) {
 
-	if nextConsumer == nil {
-		return nil, component.ErrNilNextConsumer
-	}
-
 	obsrecv, err := receiverhelper.NewObsReport(receiverhelper.ObsReportSettings{
 		ReceiverID:             settings.ID,
 		Transport:              transport,

receiver/cloudfoundryreceiver/receiver_test.go

@@ -37,19 +37,3 @@ func TestDefaultValidReceiver(t *testing.T) {
 	err = receiver.Shutdown(ctx)
 	require.NoError(t, err)
 }
-
-// Test to make sure start fails with invalid consumer
-func TestInvalidConsumer(t *testing.T) {
-	factory := NewFactory()
-	cfg := factory.CreateDefaultConfig().(*Config)
-	params := receivertest.NewNopCreateSettings()
-
-	receiver, err := newCloudFoundryReceiver(
-		params,
-		*cfg,
-		nil,
-	)
-
-	require.EqualError(t, err, "nil next Consumer")
-	require.Nil(t, receiver, "receiver creation failed")
-}

receiver/collectdreceiver/receiver.go

@@ -41,9 +41,6 @@ func newCollectdReceiver(
 	defaultAttrsPrefix string,
 	nextConsumer consumer.Metrics,
 	createSettings receiver.CreateSettings) (receiver.Metrics, error) {
-	if nextConsumer == nil {
-		return nil, component.ErrNilNextConsumer
-	}
 
 	r := &collectdReceiver{
 		logger:             logger,

receiver/datadogreceiver/receiver.go

@@ -26,9 +26,6 @@ type datadogReceiver struct {
 }
 
 func newDataDogReceiver(config *Config, nextConsumer consumer.Traces, params receiver.CreateSettings) (receiver.Traces, error) {
-	if nextConsumer == nil {
-		return nil, component.ErrNilNextConsumer
-	}
 
 	instance, err := receiverhelper.NewObsReport(receiverhelper.ObsReportSettings{LongLivedCtx: false, ReceiverID: params.ID, Transport: "http", ReceiverCreateSettings: params})
 	if err != nil {

receiver/googlecloudpubsubreceiver/factory.go

@@ -71,9 +71,6 @@ func (factory *pubsubReceiverFactory) CreateTracesReceiver(
 	cfg component.Config,
 	consumer consumer.Traces) (receiver.Traces, error) {
 
-	if consumer == nil {
-		return nil, component.ErrNilNextConsumer
-	}
 	err := cfg.(*Config).validateForTrace()
 	if err != nil {
 		return nil, err
@@ -92,9 +89,6 @@ func (factory *pubsubReceiverFactory) CreateMetricsReceiver(
 	cfg component.Config,
 	consumer consumer.Metrics) (receiver.Metrics, error) {
 
-	if consumer == nil {
-		return nil, component.ErrNilNextConsumer
-	}
 	err := cfg.(*Config).validateForMetric()
 	if err != nil {
 		return nil, err
@@ -113,9 +107,6 @@ func (factory *pubsubReceiverFactory) CreateLogsReceiver(
 	cfg component.Config,
 	consumer consumer.Logs) (receiver.Logs, error) {
 
-	if consumer == nil {
-		return nil, component.ErrNilNextConsumer
-	}
 	err := cfg.(*Config).validateForLog()
 	if err != nil {
 		return nil, err

receiver/googlecloudpubsubreceiver/factory_test.go

@@ -36,8 +36,6 @@ func TestCreateTracesReceiver(t *testing.T) {
 	tReceiver, err := factory.CreateTracesReceiver(context.Background(), params, cfg, consumertest.NewNop())
 	assert.NoError(t, err)
 	assert.NotNil(t, tReceiver, "traces receiver creation failed")
-	_, err = factory.CreateTracesReceiver(context.Background(), params, cfg, nil)
-	assert.Error(t, err)
 }
 
 func TestCreateMetricsReceiver(t *testing.T) {
@@ -49,8 +47,6 @@ func TestCreateMetricsReceiver(t *testing.T) {
 	tReceiver, err := factory.CreateMetricsReceiver(context.Background(), params, cfg, consumertest.NewNop())
 	assert.NoError(t, err)
 	assert.NotNil(t, tReceiver, "metrics receiver creation failed")
-	_, err = factory.CreateMetricsReceiver(context.Background(), params, cfg, nil)
-	assert.Error(t, err)
 }
 
 func TestCreateLogsReceiver(t *testing.T) {
@@ -62,8 +58,6 @@ func TestCreateLogsReceiver(t *testing.T) {
 	tReceiver, err := factory.CreateLogsReceiver(context.Background(), params, cfg, consumertest.NewNop())
 	assert.NoError(t, err)
 	assert.NotNil(t, tReceiver, "logs receiver creation failed")
-	_, err = factory.CreateLogsReceiver(context.Background(), params, cfg, nil)
-	assert.Error(t, err)
 }
 
 func TestEnsureReceiver(t *testing.T) {

receiver/lokireceiver/loki.go

@@ -69,10 +69,6 @@ func newLokiReceiver(conf *Config, nextConsumer consumer.Logs, settings receiver
 		return nil, err
 	}
 
-	if nextConsumer == nil {
-		return nil, component.ErrNilNextConsumer
-	}
-
 	if conf.HTTP != nil {
 		r.httpMux = http.NewServeMux()
 		r.httpMux.HandleFunc("/loki/api/v1/push", func(resp http.ResponseWriter, req *http.Request) {

receiver/opencensusreceiver/internal/ocmetrics/opencensus.go

@@ -12,7 +12,6 @@ import (
 	agentmetricspb "github.com/census-instrumentation/opencensus-proto/gen-go/agent/metrics/v1"
 	ocmetrics "github.com/census-instrumentation/opencensus-proto/gen-go/metrics/v1"
 	resourcepb "github.com/census-instrumentation/opencensus-proto/gen-go/resource/v1"
-	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/receiver"
 	"go.opentelemetry.io/collector/receiver/receiverhelper"
@@ -29,9 +28,6 @@ type Receiver struct {
 
 // New creates a new ocmetrics.Receiver reference.
 func New(nextConsumer consumer.Metrics, set receiver.CreateSettings) (*Receiver, error) {
-	if nextConsumer == nil {
-		return nil, component.ErrNilNextConsumer
-	}
 	obsrecv, err := receiverhelper.NewObsReport(receiverhelper.ObsReportSettings{
 		ReceiverID:             set.ID,
 		Transport:              receiverTransport,

receiver/opencensusreceiver/internal/octrace/opencensus.go

@@ -11,7 +11,6 @@ import (
 	commonpb "github.com/census-instrumentation/opencensus-proto/gen-go/agent/common/v1"
 	agenttracepb "github.com/census-instrumentation/opencensus-proto/gen-go/agent/trace/v1"
 	resourcepb "github.com/census-instrumentation/opencensus-proto/gen-go/resource/v1"
-	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/pdata/ptrace"
 	"go.opentelemetry.io/collector/receiver"
@@ -34,9 +33,6 @@ type Receiver struct {
 
 // New creates a new opencensus.Receiver reference.
 func New(nextConsumer consumer.Traces, set receiver.CreateSettings) (*Receiver, error) {
-	if nextConsumer == nil {
-		return nil, component.ErrNilNextConsumer
-	}
 
 	obsrecv, err := receiverhelper.NewObsReport(receiverhelper.ObsReportSettings{
 		ReceiverID:             set.ID,

receiver/otelarrowreceiver/factory.go

@@ -67,9 +67,7 @@ func createTraces(
 		return nil, err
 	}
 
-	if err = r.Unwrap().(*otelArrowReceiver).registerTraceConsumer(nextConsumer); err != nil {
-		return nil, err
-	}
+	r.Unwrap().(*otelArrowReceiver).registerTraceConsumer(nextConsumer)
 	return r, nil
 }
 
@@ -89,9 +87,7 @@ func createMetrics(
 		return nil, err
 	}
 
-	if err = r.Unwrap().(*otelArrowReceiver).registerMetricsConsumer(consumer); err != nil {
-		return nil, err
-	}
+	r.Unwrap().(*otelArrowReceiver).registerMetricsConsumer(consumer)
 	return r, nil
 }
 
@@ -111,9 +107,7 @@ func createLog(
 		return nil, err
 	}
 
-	if err = r.Unwrap().(*otelArrowReceiver).registerLogsConsumer(consumer); err != nil {
-		return nil, err
-	}
+	r.Unwrap().(*otelArrowReceiver).registerLogsConsumer(consumer)
 	return r, nil
 }
 

receiver/otelarrowreceiver/otelarrow.go

@@ -170,28 +170,16 @@ func (r *otelArrowReceiver) Shutdown(_ context.Context) error {
 	return err
 }
 
-func (r *otelArrowReceiver) registerTraceConsumer(tc consumer.Traces) error {
-	if tc == nil {
-		return component.ErrNilNextConsumer
-	}
+func (r *otelArrowReceiver) registerTraceConsumer(tc consumer.Traces) {
 	r.tracesReceiver = trace.New(tc, r.obsrepGRPC)
-	return nil
 }
 
-func (r *otelArrowReceiver) registerMetricsConsumer(mc consumer.Metrics) error {
-	if mc == nil {
-		return component.ErrNilNextConsumer
-	}
+func (r *otelArrowReceiver) registerMetricsConsumer(mc consumer.Metrics) {
 	r.metricsReceiver = metrics.New(mc, r.obsrepGRPC)
-	return nil
 }
 
-func (r *otelArrowReceiver) registerLogsConsumer(lc consumer.Logs) error {
-	if lc == nil {
-		return component.ErrNilNextConsumer
-	}
+func (r *otelArrowReceiver) registerLogsConsumer(lc consumer.Logs) {
 	r.logsReceiver = logs.New(lc, r.obsrepGRPC)
-	return nil
 }
 
 var _ arrow.Consumers = &otelArrowReceiver{}