Deprecate consumerhelper, move helpers to consumer (#5006)

Signed-off-by: Bogdan Drutu <bogdandrutu@gmail.com>

CHANGELOG.md

@@ -11,6 +11,7 @@
 ### 🚩 Deprecations 🚩
 
 - Deprecate `pdata.AttributeMap.Delete` in favor of `pdata.AttributeMap.Remove` (#4914)
+- Deprecate consumerhelper, move helpers to consumer (#5006)
 
 ### 💡 Enhancements 💡
 

consumer/consumer.go

@@ -15,9 +15,7 @@
 package consumer // import "go.opentelemetry.io/collector/consumer"
 
 import (
-	"context"
-
-	"go.opentelemetry.io/collector/model/pdata"
+	"errors"
 )
 
 // Capabilities describes the capabilities of a Processor.
@@ -34,26 +32,36 @@ type baseConsumer interface {
 	Capabilities() Capabilities
 }
 
-// Metrics is the new metrics consumer interface that receives pdata.Metrics, processes it
-// as needed, and sends it to the next processing node if any or to the destination.
-type Metrics interface {
-	baseConsumer
-	// ConsumeMetrics receives pdata.Metrics for consumption.
-	ConsumeMetrics(ctx context.Context, md pdata.Metrics) error
+var errNilFunc = errors.New("nil consumer func")
+
+type baseImpl struct {
+	capabilities Capabilities
 }
 
-// Traces is an interface that receives pdata.Traces, processes it
-// as needed, and sends it to the next processing node if any or to the destination.
-type Traces interface {
-	baseConsumer
-	// ConsumeTraces receives pdata.Traces for consumption.
-	ConsumeTraces(ctx context.Context, td pdata.Traces) error
+// Option to construct new consumers.
+type Option func(*baseImpl)
+
+// WithCapabilities overrides the default GetCapabilities function for a processor.
+// The default GetCapabilities function returns mutable capabilities.
+func WithCapabilities(capabilities Capabilities) Option {
+	return func(o *baseImpl) {
+		o.capabilities = capabilities
+	}
 }
 
-// Logs is an interface that receives pdata.Logs, processes it
-// as needed, and sends it to the next processing node if any or to the destination.
-type Logs interface {
-	baseConsumer
-	// ConsumeLogs receives pdata.Logs for consumption.
-	ConsumeLogs(ctx context.Context, ld pdata.Logs) error
+// Capabilities implementation of the base
+func (bs baseImpl) Capabilities() Capabilities {
+	return bs.capabilities
+}
+
+func newBaseImpl(options ...Option) *baseImpl {
+	bs := &baseImpl{
+		capabilities: Capabilities{MutatesData: false},
+	}
+
+	for _, op := range options {
+		op(bs)
+	}
+
+	return bs
 }

consumer/consumerhelper/common.go

@@ -15,41 +15,11 @@
 package consumerhelper // import "go.opentelemetry.io/collector/consumer/consumerhelper"
 
 import (
-	"errors"
-
 	"go.opentelemetry.io/collector/consumer"
 )
 
-var errNilFunc = errors.New("nil consumer func")
+// Deprecated: [v0.47.0] use consumer.Option
+type Option = consumer.Option
 
-type baseConsumer struct {
-	capabilities consumer.Capabilities
-}
-
-// Option applies changes to internalOptions.
-type Option func(*baseConsumer)
-
-// WithCapabilities overrides the default GetCapabilities function for a processor.
-// The default GetCapabilities function returns mutable capabilities.
-func WithCapabilities(capabilities consumer.Capabilities) Option {
-	return func(o *baseConsumer) {
-		o.capabilities = capabilities
-	}
-}
-
-// Capabilities implementation of the base Consumer.
-func (bs baseConsumer) Capabilities() consumer.Capabilities {
-	return bs.capabilities
-}
-
-func newBaseConsumer(options ...Option) *baseConsumer {
-	bs := &baseConsumer{
-		capabilities: consumer.Capabilities{MutatesData: false},
-	}
-
-	for _, op := range options {
-		op(bs)
-	}
-
-	return bs
-}
+// Deprecated: [v0.47.0] use consumer.WithCapabilities
+var WithCapabilities = consumer.WithCapabilities

consumer/consumerhelper/common_test.go

@@ -1,36 +0,0 @@
-// Copyright The OpenTelemetry Authors
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//       http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package consumerhelper
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-
-	"go.opentelemetry.io/collector/consumer"
-)
-
-func TestDefaultOptions(t *testing.T) {
-	bp := newBaseConsumer()
-	assert.Equal(t, consumer.Capabilities{MutatesData: false}, bp.Capabilities())
-}
-
-func TestWithCapabilities(t *testing.T) {
-	bpMutate := newBaseConsumer(WithCapabilities(consumer.Capabilities{MutatesData: true}))
-	assert.Equal(t, consumer.Capabilities{MutatesData: true}, bpMutate.Capabilities())
-
-	bpNotMutate := newBaseConsumer(WithCapabilities(consumer.Capabilities{MutatesData: false}))
-	assert.Equal(t, consumer.Capabilities{MutatesData: false}, bpNotMutate.Capabilities())
-}

consumer/consumerhelper/doc.go

@@ -1,17 +0,0 @@
-// Copyright The OpenTelemetry Authors
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//       http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-// Package consumerhelper defines types and functions used to create consumer
-// Logs, Metrics, and Traces.
-package consumerhelper // import "go.opentelemetry.io/collector/consumer/consumerhelper"

consumer/consumerhelper/logs.go

@@ -15,32 +15,11 @@
 package consumerhelper // import "go.opentelemetry.io/collector/consumer/consumerhelper"
 
 import (
-	"context"
-
 	"go.opentelemetry.io/collector/consumer"
-	"go.opentelemetry.io/collector/model/pdata"
 )
 
-// ConsumeLogsFunc is a helper function that is similar to ConsumeLogs.
-type ConsumeLogsFunc func(ctx context.Context, ld pdata.Logs) error
+// Deprecated: [v0.47.0] use consumer.ConsumeLogsFunc
+type ConsumeLogsFunc = consumer.ConsumeLogsFunc
 
-// ConsumeLogs calls f(ctx, ld).
-func (f ConsumeLogsFunc) ConsumeLogs(ctx context.Context, ld pdata.Logs) error {
-	return f(ctx, ld)
-}
-
-type baseLogs struct {
-	*baseConsumer
-	ConsumeLogsFunc
-}
-
-// NewLogs returns a consumer.Logs configured with the provided options.
-func NewLogs(consume ConsumeLogsFunc, options ...Option) (consumer.Logs, error) {
-	if consume == nil {
-		return nil, errNilFunc
-	}
-	return &baseLogs{
-		baseConsumer:    newBaseConsumer(options...),
-		ConsumeLogsFunc: consume,
-	}, nil
-}
+// Deprecated: [v0.47.0] use consumer.NewLogs
+var NewLogs = consumer.NewLogs

consumer/consumerhelper/metrics.go

@@ -15,32 +15,11 @@
 package consumerhelper // import "go.opentelemetry.io/collector/consumer/consumerhelper"
 
 import (
-	"context"
-
 	"go.opentelemetry.io/collector/consumer"
-	"go.opentelemetry.io/collector/model/pdata"
 )
 
-// ConsumeMetricsFunc is a helper function that is similar to ConsumeMetrics.
-type ConsumeMetricsFunc func(ctx context.Context, ld pdata.Metrics) error
+// Deprecated: [v0.47.0] use consumer.ConsumeMetricsFunc
+type ConsumeMetricsFunc = consumer.ConsumeMetricsFunc
 
-// ConsumeMetrics calls f(ctx, ld).
-func (f ConsumeMetricsFunc) ConsumeMetrics(ctx context.Context, ld pdata.Metrics) error {
-	return f(ctx, ld)
-}
-
-type baseMetrics struct {
-	*baseConsumer
-	ConsumeMetricsFunc
-}
-
-// NewMetrics returns a consumer.Metrics configured with the provided options.
-func NewMetrics(consume ConsumeMetricsFunc, options ...Option) (consumer.Metrics, error) {
-	if consume == nil {
-		return nil, errNilFunc
-	}
-	return &baseMetrics{
-		baseConsumer:       newBaseConsumer(options...),
-		ConsumeMetricsFunc: consume,
-	}, nil
-}
+// Deprecated: [v0.47.0] use consumer.NewMetrics
+var NewMetrics = consumer.NewMetrics

consumer/consumerhelper/traces.go

@@ -15,32 +15,11 @@
 package consumerhelper // import "go.opentelemetry.io/collector/consumer/consumerhelper"
 
 import (
-	"context"
-
 	"go.opentelemetry.io/collector/consumer"
-	"go.opentelemetry.io/collector/model/pdata"
 )
 
-// ConsumeTracesFunc is a helper function that is similar to ConsumeTraces.
-type ConsumeTracesFunc func(ctx context.Context, ld pdata.Traces) error
+// Deprecated: [v0.47.0] use consumer.ConsumeTracesFunc
+type ConsumeTracesFunc = consumer.ConsumeTracesFunc
 
-// ConsumeTraces calls f(ctx, ld).
-func (f ConsumeTracesFunc) ConsumeTraces(ctx context.Context, ld pdata.Traces) error {
-	return f(ctx, ld)
-}
-
-type baseTraces struct {
-	*baseConsumer
-	ConsumeTracesFunc
-}
-
-// NewTraces returns a consumer.Traces configured with the provided options.
-func NewTraces(consume ConsumeTracesFunc, options ...Option) (consumer.Traces, error) {
-	if consume == nil {
-		return nil, errNilFunc
-	}
-	return &baseTraces{
-		baseConsumer:      newBaseConsumer(options...),
-		ConsumeTracesFunc: consume,
-	}, nil
-}
+// Deprecated: [v0.47.0] use consumer.NewTraces
+var NewTraces = consumer.NewTraces

consumer/logs.go

@@ -0,0 +1,53 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package consumer // import "go.opentelemetry.io/collector/consumer"
+
+import (
+	"context"
+
+	"go.opentelemetry.io/collector/model/pdata"
+)
+
+// Logs is an interface that receives pdata.Logs, processes it
+// as needed, and sends it to the next processing node if any or to the destination.
+type Logs interface {
+	baseConsumer
+	// ConsumeLogs receives pdata.Logs for consumption.
+	ConsumeLogs(ctx context.Context, ld pdata.Logs) error
+}
+
+// ConsumeLogsFunc is a helper function that is similar to ConsumeLogs.
+type ConsumeLogsFunc func(ctx context.Context, ld pdata.Logs) error
+
+// ConsumeLogs calls f(ctx, ld).
+func (f ConsumeLogsFunc) ConsumeLogs(ctx context.Context, ld pdata.Logs) error {
+	return f(ctx, ld)
+}
+
+type baseLogs struct {
+	*baseImpl
+	ConsumeLogsFunc
+}
+
+// NewLogs returns a Logs configured with the provided options.
+func NewLogs(consume ConsumeLogsFunc, options ...Option) (Logs, error) {
+	if consume == nil {
+		return nil, errNilFunc
+	}
+	return &baseLogs{
+		baseImpl:        newBaseImpl(options...),
+		ConsumeLogsFunc: consume,
+	}, nil
+}

consumer/logs_test.go

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package consumerhelper
+package consumer
 
 import (
 	"context"
@@ -21,7 +21,6 @@ import (
 
 	"github.com/stretchr/testify/assert"
 
-	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/model/pdata"
 )
 
@@ -29,7 +28,7 @@ func TestDefaultLogs(t *testing.T) {
 	cp, err := NewLogs(func(context.Context, pdata.Logs) error { return nil })
 	assert.NoError(t, err)
 	assert.NoError(t, cp.ConsumeLogs(context.Background(), pdata.NewLogs()))
-	assert.Equal(t, consumer.Capabilities{MutatesData: false}, cp.Capabilities())
+	assert.Equal(t, Capabilities{MutatesData: false}, cp.Capabilities())
 }
 
 func TestNilFuncLogs(t *testing.T) {
@@ -40,10 +39,10 @@ func TestNilFuncLogs(t *testing.T) {
 func TestWithCapabilitiesLogs(t *testing.T) {
 	cp, err := NewLogs(
 		func(context.Context, pdata.Logs) error { return nil },
-		WithCapabilities(consumer.Capabilities{MutatesData: true}))
+		WithCapabilities(Capabilities{MutatesData: true}))
 	assert.NoError(t, err)
 	assert.NoError(t, cp.ConsumeLogs(context.Background(), pdata.NewLogs()))
-	assert.Equal(t, consumer.Capabilities{MutatesData: true}, cp.Capabilities())
+	assert.Equal(t, Capabilities{MutatesData: true}, cp.Capabilities())
 }
 
 func TestConsumeLogs(t *testing.T) {

consumer/metrics.go

@@ -0,0 +1,53 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package consumer // import "go.opentelemetry.io/collector/consumer"
+
+import (
+	"context"
+
+	"go.opentelemetry.io/collector/model/pdata"
+)
+
+// Metrics is the new metrics consumer interface that receives pdata.Metrics, processes it
+// as needed, and sends it to the next processing node if any or to the destination.
+type Metrics interface {
+	baseConsumer
+	// ConsumeMetrics receives pdata.Metrics for consumption.
+	ConsumeMetrics(ctx context.Context, md pdata.Metrics) error
+}
+
+// ConsumeMetricsFunc is a helper function that is similar to ConsumeMetrics.
+type ConsumeMetricsFunc func(ctx context.Context, ld pdata.Metrics) error
+
+// ConsumeMetrics calls f(ctx, ld).
+func (f ConsumeMetricsFunc) ConsumeMetrics(ctx context.Context, ld pdata.Metrics) error {
+	return f(ctx, ld)
+}
+
+type baseMetrics struct {
+	*baseImpl
+	ConsumeMetricsFunc
+}
+
+// NewMetrics returns a Metrics configured with the provided options.
+func NewMetrics(consume ConsumeMetricsFunc, options ...Option) (Metrics, error) {
+	if consume == nil {
+		return nil, errNilFunc
+	}
+	return &baseMetrics{
+		baseImpl:           newBaseImpl(options...),
+		ConsumeMetricsFunc: consume,
+	}, nil
+}

consumer/metrics_test.go

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package consumerhelper
+package consumer
 
 import (
 	"context"
@@ -21,7 +21,6 @@ import (
 
 	"github.com/stretchr/testify/assert"
 
-	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/model/pdata"
 )
 
@@ -29,7 +28,7 @@ func TestDefaultMetrics(t *testing.T) {
 	cp, err := NewMetrics(func(context.Context, pdata.Metrics) error { return nil })
 	assert.NoError(t, err)
 	assert.NoError(t, cp.ConsumeMetrics(context.Background(), pdata.NewMetrics()))
-	assert.Equal(t, consumer.Capabilities{MutatesData: false}, cp.Capabilities())
+	assert.Equal(t, Capabilities{MutatesData: false}, cp.Capabilities())
 }
 
 func TestNilFuncMetrics(t *testing.T) {
@@ -40,10 +39,10 @@ func TestNilFuncMetrics(t *testing.T) {
 func TestWithCapabilitiesMetrics(t *testing.T) {
 	cp, err := NewMetrics(
 		func(context.Context, pdata.Metrics) error { return nil },
-		WithCapabilities(consumer.Capabilities{MutatesData: true}))
+		WithCapabilities(Capabilities{MutatesData: true}))
 	assert.NoError(t, err)
 	assert.NoError(t, cp.ConsumeMetrics(context.Background(), pdata.NewMetrics()))
-	assert.Equal(t, consumer.Capabilities{MutatesData: true}, cp.Capabilities())
+	assert.Equal(t, Capabilities{MutatesData: true}, cp.Capabilities())
 }
 
 func TestConsumeMetrics(t *testing.T) {

consumer/traces.go

@@ -0,0 +1,53 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package consumer // import "go.opentelemetry.io/collector/consumer"
+
+import (
+	"context"
+
+	"go.opentelemetry.io/collector/model/pdata"
+)
+
+// Traces is an interface that receives pdata.Traces, processes it
+// as needed, and sends it to the next processing node if any or to the destination.
+type Traces interface {
+	baseConsumer
+	// ConsumeTraces receives pdata.Traces for consumption.
+	ConsumeTraces(ctx context.Context, td pdata.Traces) error
+}
+
+// ConsumeTracesFunc is a helper function that is similar to ConsumeTraces.
+type ConsumeTracesFunc func(ctx context.Context, ld pdata.Traces) error
+
+// ConsumeTraces calls f(ctx, ld).
+func (f ConsumeTracesFunc) ConsumeTraces(ctx context.Context, ld pdata.Traces) error {
+	return f(ctx, ld)
+}
+
+type baseTraces struct {
+	*baseImpl
+	ConsumeTracesFunc
+}
+
+// NewTraces returns a Traces configured with the provided options.
+func NewTraces(consume ConsumeTracesFunc, options ...Option) (Traces, error) {
+	if consume == nil {
+		return nil, errNilFunc
+	}
+	return &baseTraces{
+		baseImpl:          newBaseImpl(options...),
+		ConsumeTracesFunc: consume,
+	}, nil
+}

consumer/traces_test.go

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package consumerhelper
+package consumer
 
 import (
 	"context"
@@ -21,7 +21,6 @@ import (
 
 	"github.com/stretchr/testify/assert"
 
-	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/model/pdata"
 )
 
@@ -29,7 +28,7 @@ func TestDefaultTraces(t *testing.T) {
 	cp, err := NewTraces(func(context.Context, pdata.Traces) error { return nil })
 	assert.NoError(t, err)
 	assert.NoError(t, cp.ConsumeTraces(context.Background(), pdata.NewTraces()))
-	assert.Equal(t, consumer.Capabilities{MutatesData: false}, cp.Capabilities())
+	assert.Equal(t, Capabilities{MutatesData: false}, cp.Capabilities())
 }
 
 func TestNilFuncTraces(t *testing.T) {
@@ -40,10 +39,10 @@ func TestNilFuncTraces(t *testing.T) {
 func TestWithCapabilitiesTraces(t *testing.T) {
 	cp, err := NewTraces(
 		func(context.Context, pdata.Traces) error { return nil },
-		WithCapabilities(consumer.Capabilities{MutatesData: true}))
+		WithCapabilities(Capabilities{MutatesData: true}))
 	assert.NoError(t, err)
 	assert.NoError(t, cp.ConsumeTraces(context.Background(), pdata.NewTraces()))
-	assert.Equal(t, consumer.Capabilities{MutatesData: true}, cp.Capabilities())
+	assert.Equal(t, Capabilities{MutatesData: true}, cp.Capabilities())
 }
 
 func TestConsumeTraces(t *testing.T) {

exporter/exporterhelper/common.go

@@ -21,7 +21,6 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/consumer"
-	"go.opentelemetry.io/collector/consumer/consumerhelper"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal"
 	"go.opentelemetry.io/collector/obsreport"
 )
@@ -89,7 +88,7 @@ func (req *baseRequest) OnProcessingFinished() {
 type baseSettings struct {
 	component.StartFunc
 	component.ShutdownFunc
-	consumerOptions []consumerhelper.Option
+	consumerOptions []consumer.Option
 	TimeoutSettings
 	QueueSettings
 	RetrySettings
@@ -161,7 +160,7 @@ func WithQueue(queueSettings QueueSettings) Option {
 // TODO: Verify if we can change the default to be mutable as we do for processors.
 func WithCapabilities(capabilities consumer.Capabilities) Option {
 	return func(o *baseSettings) {
-		o.consumerOptions = append(o.consumerOptions, consumerhelper.WithCapabilities(capabilities))
+		o.consumerOptions = append(o.consumerOptions, consumer.WithCapabilities(capabilities))
 	}
 }
 

exporter/exporterhelper/common_test.go

@@ -26,7 +26,7 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/config"
-	"go.opentelemetry.io/collector/consumer/consumerhelper"
+	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal"
 	"go.opentelemetry.io/collector/model/pdata"
 )
@@ -70,7 +70,7 @@ func checkStatus(t *testing.T, sd sdktrace.ReadOnlySpan, err error) {
 	}
 }
 
-func nopTracePusher() consumerhelper.ConsumeTracesFunc {
+func nopTracePusher() consumer.ConsumeTracesFunc {
 	return func(ctx context.Context, ld pdata.Traces) error {
 		return nil
 	}

exporter/exporterhelper/logs.go

@@ -22,7 +22,6 @@ import (
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/consumer/consumererror"
-	"go.opentelemetry.io/collector/consumer/consumerhelper"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal"
 	"go.opentelemetry.io/collector/model/otlp"
 	"go.opentelemetry.io/collector/model/pdata"
@@ -34,10 +33,10 @@ var logsUnmarshaler = otlp.NewProtobufLogsUnmarshaler()
 type logsRequest struct {
 	baseRequest
 	ld     pdata.Logs
-	pusher consumerhelper.ConsumeLogsFunc
+	pusher consumer.ConsumeLogsFunc
 }
 
-func newLogsRequest(ctx context.Context, ld pdata.Logs, pusher consumerhelper.ConsumeLogsFunc) request {
+func newLogsRequest(ctx context.Context, ld pdata.Logs, pusher consumer.ConsumeLogsFunc) request {
 	return &logsRequest{
 		baseRequest: baseRequest{ctx: ctx},
 		ld:          ld,
@@ -45,7 +44,7 @@ func newLogsRequest(ctx context.Context, ld pdata.Logs, pusher consumerhelper.Co
 	}
 }
 
-func newLogsRequestUnmarshalerFunc(pusher consumerhelper.ConsumeLogsFunc) internal.RequestUnmarshaler {
+func newLogsRequestUnmarshalerFunc(pusher consumer.ConsumeLogsFunc) internal.RequestUnmarshaler {
 	return func(bytes []byte) (internal.PersistentRequest, error) {
 		logs, err := logsUnmarshaler.UnmarshalLogs(bytes)
 		if err != nil {
@@ -84,7 +83,7 @@ type logsExporter struct {
 func NewLogsExporter(
 	cfg config.Exporter,
 	set component.ExporterCreateSettings,
-	pusher consumerhelper.ConsumeLogsFunc,
+	pusher consumer.ConsumeLogsFunc,
 	options ...Option,
 ) (component.LogsExporter, error) {
 	if cfg == nil {
@@ -108,7 +107,7 @@ func NewLogsExporter(
 		}
 	})
 
-	lc, err := consumerhelper.NewLogs(func(ctx context.Context, ld pdata.Logs) error {
+	lc, err := consumer.NewLogs(func(ctx context.Context, ld pdata.Logs) error {
 		req := newLogsRequest(ctx, ld, pusher)
 		err := be.sender.send(req)
 		if errors.Is(err, errSendingQueueIsFull) {

exporter/exporterhelper/logs_test.go

@@ -31,7 +31,6 @@ import (
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/consumer/consumererror"
-	"go.opentelemetry.io/collector/consumer/consumerhelper"
 	"go.opentelemetry.io/collector/internal/obsreportconfig/obsmetrics"
 	"go.opentelemetry.io/collector/internal/testdata"
 	"go.opentelemetry.io/collector/model/pdata"
@@ -199,7 +198,7 @@ func TestLogsExporter_WithShutdown_ReturnError(t *testing.T) {
 	assert.Equal(t, le.Shutdown(context.Background()), want)
 }
 
-func newPushLogsData(retError error) consumerhelper.ConsumeLogsFunc {
+func newPushLogsData(retError error) consumer.ConsumeLogsFunc {
 	return func(ctx context.Context, td pdata.Logs) error {
 		return retError
 	}

exporter/exporterhelper/metrics.go

@@ -22,7 +22,6 @@ import (
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/consumer/consumererror"
-	"go.opentelemetry.io/collector/consumer/consumerhelper"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal"
 	"go.opentelemetry.io/collector/model/otlp"
 	"go.opentelemetry.io/collector/model/pdata"
@@ -34,10 +33,10 @@ var metricsUnmarshaler = otlp.NewProtobufMetricsUnmarshaler()
 type metricsRequest struct {
 	baseRequest
 	md     pdata.Metrics
-	pusher consumerhelper.ConsumeMetricsFunc
+	pusher consumer.ConsumeMetricsFunc
 }
 
-func newMetricsRequest(ctx context.Context, md pdata.Metrics, pusher consumerhelper.ConsumeMetricsFunc) request {
+func newMetricsRequest(ctx context.Context, md pdata.Metrics, pusher consumer.ConsumeMetricsFunc) request {
 	return &metricsRequest{
 		baseRequest: baseRequest{ctx: ctx},
 		md:          md,
@@ -45,7 +44,7 @@ func newMetricsRequest(ctx context.Context, md pdata.Metrics, pusher consumerhel
 	}
 }
 
-func newMetricsRequestUnmarshalerFunc(pusher consumerhelper.ConsumeMetricsFunc) internal.RequestUnmarshaler {
+func newMetricsRequestUnmarshalerFunc(pusher consumer.ConsumeMetricsFunc) internal.RequestUnmarshaler {
 	return func(bytes []byte) (internal.PersistentRequest, error) {
 		metrics, err := metricsUnmarshaler.UnmarshalMetrics(bytes)
 		if err != nil {
@@ -85,7 +84,7 @@ type metricsExporter struct {
 func NewMetricsExporter(
 	cfg config.Exporter,
 	set component.ExporterCreateSettings,
-	pusher consumerhelper.ConsumeMetricsFunc,
+	pusher consumer.ConsumeMetricsFunc,
 	options ...Option,
 ) (component.MetricsExporter, error) {
 	if cfg == nil {
@@ -109,7 +108,7 @@ func NewMetricsExporter(
 		}
 	})
 
-	mc, err := consumerhelper.NewMetrics(func(ctx context.Context, md pdata.Metrics) error {
+	mc, err := consumer.NewMetrics(func(ctx context.Context, md pdata.Metrics) error {
 		req := newMetricsRequest(ctx, md, pusher)
 		err := be.sender.send(req)
 		if errors.Is(err, errSendingQueueIsFull) {

exporter/exporterhelper/metrics_test.go

@@ -31,7 +31,6 @@ import (
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/consumer/consumererror"
-	"go.opentelemetry.io/collector/consumer/consumerhelper"
 	"go.opentelemetry.io/collector/internal/obsreportconfig/obsmetrics"
 	"go.opentelemetry.io/collector/internal/testdata"
 	"go.opentelemetry.io/collector/model/pdata"
@@ -200,7 +199,7 @@ func TestMetricsExporter_WithShutdown_ReturnError(t *testing.T) {
 	assert.Equal(t, want, me.Shutdown(context.Background()))
 }
 
-func newPushMetricsData(retError error) consumerhelper.ConsumeMetricsFunc {
+func newPushMetricsData(retError error) consumer.ConsumeMetricsFunc {
 	return func(ctx context.Context, td pdata.Metrics) error {
 		return retError
 	}

exporter/exporterhelper/traces.go

@@ -22,7 +22,6 @@ import (
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/consumer/consumererror"
-	"go.opentelemetry.io/collector/consumer/consumerhelper"
 	"go.opentelemetry.io/collector/exporter/exporterhelper/internal"
 	"go.opentelemetry.io/collector/model/otlp"
 	"go.opentelemetry.io/collector/model/pdata"
@@ -34,10 +33,10 @@ var tracesUnmarshaler = otlp.NewProtobufTracesUnmarshaler()
 type tracesRequest struct {
 	baseRequest
 	td     pdata.Traces
-	pusher consumerhelper.ConsumeTracesFunc
+	pusher consumer.ConsumeTracesFunc
 }
 
-func newTracesRequest(ctx context.Context, td pdata.Traces, pusher consumerhelper.ConsumeTracesFunc) request {
+func newTracesRequest(ctx context.Context, td pdata.Traces, pusher consumer.ConsumeTracesFunc) request {
 	return &tracesRequest{
 		baseRequest: baseRequest{ctx: ctx},
 		td:          td,
@@ -45,7 +44,7 @@ func newTracesRequest(ctx context.Context, td pdata.Traces, pusher consumerhelpe
 	}
 }
 
-func newTraceRequestUnmarshalerFunc(pusher consumerhelper.ConsumeTracesFunc) internal.RequestUnmarshaler {
+func newTraceRequestUnmarshalerFunc(pusher consumer.ConsumeTracesFunc) internal.RequestUnmarshaler {
 	return func(bytes []byte) (internal.PersistentRequest, error) {
 		traces, err := tracesUnmarshaler.UnmarshalTraces(bytes)
 		if err != nil {
@@ -85,7 +84,7 @@ type traceExporter struct {
 func NewTracesExporter(
 	cfg config.Exporter,
 	set component.ExporterCreateSettings,
-	pusher consumerhelper.ConsumeTracesFunc,
+	pusher consumer.ConsumeTracesFunc,
 	options ...Option,
 ) (component.TracesExporter, error) {
 
@@ -110,7 +109,7 @@ func NewTracesExporter(
 		}
 	})
 
-	tc, err := consumerhelper.NewTraces(func(ctx context.Context, td pdata.Traces) error {
+	tc, err := consumer.NewTraces(func(ctx context.Context, td pdata.Traces) error {
 		req := newTracesRequest(ctx, td, pusher)
 		err := be.sender.send(req)
 		if errors.Is(err, errSendingQueueIsFull) {

exporter/exporterhelper/traces_test.go

@@ -31,7 +31,6 @@ import (
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/consumer/consumererror"
-	"go.opentelemetry.io/collector/consumer/consumerhelper"
 	"go.opentelemetry.io/collector/internal/obsreportconfig/obsmetrics"
 	"go.opentelemetry.io/collector/internal/testdata"
 	"go.opentelemetry.io/collector/model/pdata"
@@ -200,7 +199,7 @@ func TestTracesExporter_WithShutdown_ReturnError(t *testing.T) {
 	assert.Equal(t, te.Shutdown(context.Background()), want)
 }
 
-func newTraceDataPusher(retError error) consumerhelper.ConsumeTracesFunc {
+func newTraceDataPusher(retError error) consumer.ConsumeTracesFunc {
 	return func(ctx context.Context, td pdata.Traces) error {
 		return retError
 	}

processor/processorhelper/logs.go

@@ -24,7 +24,6 @@ import (
 	"go.opentelemetry.io/collector/component/componenterror"
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/consumer"
-	"go.opentelemetry.io/collector/consumer/consumerhelper"
 	"go.opentelemetry.io/collector/model/pdata"
 )
 
@@ -56,7 +55,7 @@ func NewLogsProcessor(
 
 	eventOptions := spanAttributes(cfg.ID())
 	bs := fromOptions(options)
-	logsConsumer, err := consumerhelper.NewLogs(func(ctx context.Context, ld pdata.Logs) error {
+	logsConsumer, err := consumer.NewLogs(func(ctx context.Context, ld pdata.Logs) error {
 		span := trace.SpanFromContext(ctx)
 		span.AddEvent("Start processing.", eventOptions)
 		var err error

processor/processorhelper/metrics.go

@@ -24,7 +24,6 @@ import (
 	"go.opentelemetry.io/collector/component/componenterror"
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/consumer"
-	"go.opentelemetry.io/collector/consumer/consumerhelper"
 	"go.opentelemetry.io/collector/model/pdata"
 )
 
@@ -56,7 +55,7 @@ func NewMetricsProcessor(
 
 	eventOptions := spanAttributes(cfg.ID())
 	bs := fromOptions(options)
-	metricsConsumer, err := consumerhelper.NewMetrics(func(ctx context.Context, md pdata.Metrics) error {
+	metricsConsumer, err := consumer.NewMetrics(func(ctx context.Context, md pdata.Metrics) error {
 		span := trace.SpanFromContext(ctx)
 		span.AddEvent("Start processing.", eventOptions)
 		var err error

processor/processorhelper/processor.go

@@ -23,7 +23,6 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/consumer"
-	"go.opentelemetry.io/collector/consumer/consumerhelper"
 	"go.opentelemetry.io/collector/internal/obsreportconfig/obsmetrics"
 )
 
@@ -55,21 +54,21 @@ func WithShutdown(shutdown component.ShutdownFunc) Option {
 // The default GetCapabilities function returns mutable capabilities.
 func WithCapabilities(capabilities consumer.Capabilities) Option {
 	return func(o *baseSettings) {
-		o.consumerOptions = append(o.consumerOptions, consumerhelper.WithCapabilities(capabilities))
+		o.consumerOptions = append(o.consumerOptions, consumer.WithCapabilities(capabilities))
 	}
 }
 
 type baseSettings struct {
 	component.StartFunc
 	component.ShutdownFunc
-	consumerOptions []consumerhelper.Option
+	consumerOptions []consumer.Option
 }
 
 // fromOptions returns the internal settings starting from the default and applying all options.
 func fromOptions(options []Option) *baseSettings {
 	// Start from the default options:
 	opts := &baseSettings{
-		consumerOptions: []consumerhelper.Option{consumerhelper.WithCapabilities(consumer.Capabilities{MutatesData: true})},
+		consumerOptions: []consumer.Option{consumer.WithCapabilities(consumer.Capabilities{MutatesData: true})},
 	}
 
 	for _, op := range options {

processor/processorhelper/traces.go

@@ -24,7 +24,6 @@ import (
 	"go.opentelemetry.io/collector/component/componenterror"
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/consumer"
-	"go.opentelemetry.io/collector/consumer/consumerhelper"
 	"go.opentelemetry.io/collector/model/pdata"
 )
 
@@ -56,7 +55,7 @@ func NewTracesProcessor(
 
 	eventOptions := spanAttributes(cfg.ID())
 	bs := fromOptions(options)
-	traceConsumer, err := consumerhelper.NewTraces(func(ctx context.Context, td pdata.Traces) error {
+	traceConsumer, err := consumer.NewTraces(func(ctx context.Context, td pdata.Traces) error {
 		span := trace.SpanFromContext(ctx)
 		span.AddEvent("Start processing.", eventOptions)
 		var err error