[receiver/kafkareceiver] Add otlp_json support in kafka receiver

receiver/kafkareceiver/README.md

@@ -31,6 +31,7 @@ The following settings can be optionally configured:
   Only one telemetry type may be used for a given topic.
 - `encoding` (default = otlp_proto): The encoding of the payload received from kafka. Available encodings:
   - `otlp_proto`: the payload is deserialized to `ExportTraceServiceRequest`, `ExportLogsServiceRequest` or `ExportMetricsServiceRequest` respectively.
+  - `otlp_json`: The payload is deserialized into OpenTelemetry traces using JSON format.This encoding uses the `ptrace.JSONUnmarshaler` to convert the JSON-encoded trace data into the internal OpenTelemetry format. The expected JSON structure includes resourceSpans containing trace data.
   - `jaeger_proto`: the payload is deserialized to a single Jaeger proto `Span`.
   - `jaeger_json`: the payload is deserialized to a single Jaeger JSON Span using `jsonpb`.
   - `zipkin_proto`: the payload is deserialized into a list of Zipkin proto spans.

receiver/kafkareceiver/otlp_json_unmarshaler.go

@@ -0,0 +1,32 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package kafkareceiver // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kafkareceiver"
+
+import (
+    "go.opentelemetry.io/collector/pdata/ptrace"
+)
+
+const (
+    otlpJSONEncoding = "otlp_json"
+)
+
+// otlpJSONTracesUnmarshaler unmarshals OTLP JSON-encoded traces.
+type otlpJSONTracesUnmarshaler struct {
+    ptrace.Unmarshaler
+}
+
+func (o otlpJSONTracesUnmarshaler) Unmarshal(buf []byte) (ptrace.Traces, error) {
+    return o.Unmarshaler.UnmarshalTraces(buf)
+}
+
+func (o otlpJSONTracesUnmarshaler) Encoding() string {
+    return otlpJSONEncoding
+}
+
+// newOTLPJSONTracesUnmarshaler creates a new OTLP JSON traces unmarshaler.
+func newOTLPJSONTracesUnmarshaler() TracesUnmarshaler {
+    return &otlpJSONTracesUnmarshaler{
+        Unmarshaler: &ptrace.JSONUnmarshaler{},
+    }
+}

receiver/kafkareceiver/otlp_json_unmarshaler_test.go

@@ -0,0 +1,27 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package kafkareceiver
+
+import (
+    "testing"
+
+    "github.com/stretchr/testify/assert"
+    "go.opentelemetry.io/collector/pdata/ptrace"
+)
+
+func TestNewOTLPJSONTracesUnmarshaler(t *testing.T) {
+    t.Parallel()
+    um := newOTLPJSONTracesUnmarshaler()
+    assert.Equal(t, otlpJSONEncoding, um.Encoding())
+}
+
+func TestOTLPJSONTracesUnmarshalerReturnType(t *testing.T) {
+    t.Parallel()
+    um := newOTLPJSONTracesUnmarshaler()
+    json := `{"resourceSpans":[{"resource":{},"scopeSpans":[{"scope":{},"spans":[]}]}]}`
+    unmarshaledTraces, err := um.Unmarshal([]byte(json))
+    assert.NoError(t, err)
+    var expectedType ptrace.Traces
+    assert.IsType(t, expectedType, unmarshaledTraces)
+}

receiver/kafkareceiver/unmarshaler.go

@@ -50,13 +50,15 @@ type LogsUnmarshalerWithEnc interface {
 // defaultTracesUnmarshalers returns map of supported encodings with TracesUnmarshaler.
 func defaultTracesUnmarshalers() map[string]TracesUnmarshaler {
 	otlpPb := newPdataTracesUnmarshaler(&ptrace.ProtoUnmarshaler{}, defaultEncoding)
+	otlpJSON := newOTLPJSONTracesUnmarshaler()
 	jaegerProto := jaegerProtoSpanUnmarshaler{}
 	jaegerJSON := jaegerJSONSpanUnmarshaler{}
 	zipkinProto := newPdataTracesUnmarshaler(zipkinv2.NewProtobufTracesUnmarshaler(false, false), "zipkin_proto")
 	zipkinJSON := newPdataTracesUnmarshaler(zipkinv2.NewJSONTracesUnmarshaler(false), "zipkin_json")
 	zipkinThrift := newPdataTracesUnmarshaler(zipkinv1.NewThriftTracesUnmarshaler(), "zipkin_thrift")
 	return map[string]TracesUnmarshaler{
 		otlpPb.Encoding():       otlpPb,
+		otlpJSON.Encoding(): otlpJSON,
 		jaegerProto.Encoding():  jaegerProto,
 		jaegerJSON.Encoding():   jaegerJSON,
 		zipkinProto.Encoding():  zipkinProto,

receiver/kafkareceiver/unmarshaler_test.go

@@ -14,6 +14,7 @@ import (
 func TestDefaultTracesUnMarshaler(t *testing.T) {
 	expectedEncodings := []string{
 		"otlp_proto",
+		"otlp_json",
 		"jaeger_proto",
 		"jaeger_json",
 		"zipkin_proto",