Use the new marshaler/unmarshaler in Kafka receiver (#3402)

Signed-off-by: Bogdan Drutu <[email protected]>

receiver/kafkareceiver/factory_test.go

@@ -73,7 +73,7 @@ func TestWithTracesUnmarshalers(t *testing.T) {
 		require.NotNil(t, receiver)
 	})
 	t.Run("default_encoding", func(t *testing.T) {
-		cfg.Encoding = new(otlpTracesPbUnmarshaler).Encoding()
+		cfg.Encoding = defaultEncoding
 		receiver, err := f.CreateTracesReceiver(context.Background(), componenttest.NewNopReceiverCreateSettings(), cfg, nil)
 		require.NoError(t, err)
 		assert.NotNil(t, receiver)
@@ -117,7 +117,7 @@ func TestWithLogsUnmarshalers(t *testing.T) {
 		require.NotNil(t, exporter)
 	})
 	t.Run("default_encoding", func(t *testing.T) {
-		cfg.Encoding = new(otlpLogsPbUnmarshaler).Encoding()
+		cfg.Encoding = defaultEncoding
 		exporter, err := f.CreateLogsReceiver(context.Background(), componenttest.NewNopReceiverCreateSettings(), cfg, nil)
 		require.NoError(t, err)
 		assert.NotNil(t, exporter)

receiver/kafkareceiver/kafka_receiver_test.go

@@ -34,6 +34,7 @@ import (
 	"go.opentelemetry.io/collector/consumer/consumertest"
 	"go.opentelemetry.io/collector/consumer/pdata"
 	"go.opentelemetry.io/collector/exporter/kafkaexporter"
+	"go.opentelemetry.io/collector/internal/otlp"
 	"go.opentelemetry.io/collector/internal/testdata"
 	"go.opentelemetry.io/collector/obsreport"
 )
@@ -129,7 +130,7 @@ func TestTracesConsumerGroupHandler(t *testing.T) {
 	defer view.Unregister(views...)
 
 	c := tracesConsumerGroupHandler{
-		unmarshaler:  &otlpTracesPbUnmarshaler{},
+		unmarshaler:  newPdataTracesUnmarshaler(otlp.NewProtobufTracesUnmarshaler(), defaultEncoding),
 		logger:       zap.NewNop(),
 		ready:        make(chan bool),
 		nextConsumer: consumertest.NewNop(),
@@ -171,7 +172,7 @@ func TestTracesConsumerGroupHandler(t *testing.T) {
 
 func TestTracesConsumerGroupHandler_error_unmarshal(t *testing.T) {
 	c := tracesConsumerGroupHandler{
-		unmarshaler:  &otlpTracesPbUnmarshaler{},
+		unmarshaler:  newPdataTracesUnmarshaler(otlp.NewProtobufTracesUnmarshaler(), defaultEncoding),
 		logger:       zap.NewNop(),
 		ready:        make(chan bool),
 		nextConsumer: consumertest.NewNop(),
@@ -196,7 +197,7 @@ func TestTracesConsumerGroupHandler_error_unmarshal(t *testing.T) {
 func TestTracesConsumerGroupHandler_error_nextConsumer(t *testing.T) {
 	consumerError := errors.New("failed to consumer")
 	c := tracesConsumerGroupHandler{
-		unmarshaler:  &otlpTracesPbUnmarshaler{},
+		unmarshaler:  newPdataTracesUnmarshaler(otlp.NewProtobufTracesUnmarshaler(), defaultEncoding),
 		logger:       zap.NewNop(),
 		ready:        make(chan bool),
 		nextConsumer: consumertest.NewErr(consumerError),
@@ -314,7 +315,7 @@ func TestLogsConsumerGroupHandler(t *testing.T) {
 	defer view.Unregister(views...)
 
 	c := logsConsumerGroupHandler{
-		unmarshaler:  &otlpLogsPbUnmarshaler{},
+		unmarshaler:  newPdataLogsUnmarshaler(otlp.NewProtobufLogsUnmarshaler(), defaultEncoding),
 		logger:       zap.NewNop(),
 		ready:        make(chan bool),
 		nextConsumer: consumertest.NewNop(),
@@ -356,7 +357,7 @@ func TestLogsConsumerGroupHandler(t *testing.T) {
 
 func TestLogsConsumerGroupHandler_error_unmarshal(t *testing.T) {
 	c := logsConsumerGroupHandler{
-		unmarshaler:  &otlpLogsPbUnmarshaler{},
+		unmarshaler:  newPdataLogsUnmarshaler(otlp.NewProtobufLogsUnmarshaler(), defaultEncoding),
 		logger:       zap.NewNop(),
 		ready:        make(chan bool),
 		nextConsumer: consumertest.NewNop(),
@@ -381,7 +382,7 @@ func TestLogsConsumerGroupHandler_error_unmarshal(t *testing.T) {
 func TestLogsConsumerGroupHandler_error_nextConsumer(t *testing.T) {
 	consumerError := errors.New("failed to consumer")
 	c := logsConsumerGroupHandler{
-		unmarshaler:  &otlpLogsPbUnmarshaler{},
+		unmarshaler:  newPdataLogsUnmarshaler(otlp.NewProtobufLogsUnmarshaler(), defaultEncoding),
 		logger:       zap.NewNop(),
 		ready:        make(chan bool),
 		nextConsumer: consumertest.NewErr(consumerError),

receiver/kafkareceiver/pdata_unmarshaler.go

@@ -0,0 +1,51 @@
+// 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 kafkareceiver
+
+import (
+	"go.opentelemetry.io/collector/internal/model"
+)
+
+type pdataLogsUnmarshaler struct {
+	model.LogsUnmarshaler
+	encoding string
+}
+
+func (p pdataLogsUnmarshaler) Encoding() string {
+	return p.encoding
+}
+
+func newPdataLogsUnmarshaler(unmarshaler model.LogsUnmarshaler, encoding string) LogsUnmarshaler {
+	return pdataLogsUnmarshaler{
+		LogsUnmarshaler: unmarshaler,
+		encoding:        encoding,
+	}
+}
+
+type pdataTracesUnmarshaler struct {
+	model.TracesUnmarshaler
+	encoding string
+}
+
+func (p pdataTracesUnmarshaler) Encoding() string {
+	return p.encoding
+}
+
+func newPdataTracesUnmarshaler(unmarshaler model.TracesUnmarshaler, encoding string) TracesUnmarshaler {
+	return pdataTracesUnmarshaler{
+		TracesUnmarshaler: unmarshaler,
+		encoding:          encoding,
+	}
+}

receiver/kafkareceiver/pdata_unmarshaler_test.go

@@ -0,0 +1,33 @@
+// 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 kafkareceiver
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"go.opentelemetry.io/collector/internal/otlp"
+)
+
+func TestNewPdataTracesUnmarshaler(t *testing.T) {
+	um := newPdataTracesUnmarshaler(otlp.NewProtobufTracesUnmarshaler(), "test")
+	assert.Equal(t, "test", um.Encoding())
+}
+
+func TestNewPdataLogsUnmarshaler(t *testing.T) {
+	um := newPdataLogsUnmarshaler(otlp.NewProtobufLogsUnmarshaler(), "test")
+	assert.Equal(t, "test", um.Encoding())
+}

receiver/kafkareceiver/unmarshaler.go

@@ -16,6 +16,7 @@ package kafkareceiver
 
 import (
 	"go.opentelemetry.io/collector/consumer/pdata"
+	"go.opentelemetry.io/collector/internal/otlp"
 )
 
 // TracesUnmarshaler deserializes the message body.
@@ -38,14 +39,14 @@ type LogsUnmarshaler interface {
 
 // defaultTracesUnmarshalers returns map of supported encodings with TracesUnmarshaler.
 func defaultTracesUnmarshalers() map[string]TracesUnmarshaler {
-	otlp := &otlpTracesPbUnmarshaler{}
+	otlpPb := newPdataTracesUnmarshaler(otlp.NewProtobufTracesUnmarshaler(), defaultEncoding)
 	jaegerProto := jaegerProtoSpanUnmarshaler{}
 	jaegerJSON := jaegerJSONSpanUnmarshaler{}
 	zipkinProto := zipkinProtoSpanUnmarshaler{}
 	zipkinJSON := zipkinJSONSpanUnmarshaler{}
 	zipkinThrift := zipkinThriftSpanUnmarshaler{}
 	return map[string]TracesUnmarshaler{
-		otlp.Encoding():         otlp,
+		otlpPb.Encoding():       otlpPb,
 		jaegerProto.Encoding():  jaegerProto,
 		jaegerJSON.Encoding():   jaegerJSON,
 		zipkinProto.Encoding():  zipkinProto,
@@ -55,8 +56,8 @@ func defaultTracesUnmarshalers() map[string]TracesUnmarshaler {
 }
 
 func defaultLogsUnmarshalers() map[string]LogsUnmarshaler {
-	otlp := &otlpLogsPbUnmarshaler{}
+	otlpPb := newPdataLogsUnmarshaler(otlp.NewProtobufLogsUnmarshaler(), defaultEncoding)
 	return map[string]LogsUnmarshaler{
-		otlp.Encoding(): otlp,
+		otlpPb.Encoding(): otlpPb,
 	}
 }