Add Kafka exporter and receiver configuration

cmd/jaeger/collector-with-kafka.yaml

@@ -0,0 +1,28 @@
+receivers:
+  otlp:
+    protocols:
+      grpc:
+      http:
+  jaeger:
+    protocols:
+      grpc:
+      thrift_binary:
+      thrift_compact:
+      thrift_http:
+
+processors:
+  batch:
+
+exporters:
+  kafka:
+    brokers:
+      - localhost:9092
+    topic: "jaeger-spans"
+    encoding: otlp_proto
+
+service:
+  pipelines:
+    traces:
+      receivers: [otlp, jaeger]
+      processors: [batch]
+      exporters: [kafka]

cmd/jaeger/ingester-remote-backend-storage.yaml

@@ -0,0 +1,50 @@
+service:
+  extensions: [jaeger_storage, jaeger_query]
+  pipelines:
+    traces:
+      receivers: [kafka, otlp, jaeger, zipkin]
+      processors: [batch]
+      exporters: [jaeger_storage_exporter]
+  telemetry:
+    metrics:
+      address: 0.0.0.0:8889
+    logs:
+      level: debug
+
+extensions:
+  jaeger_query:
+    trace_storage: some_storage
+
+  jaeger_storage:
+    backends:
+      some_storage:
+        memory:
+          max_traces: 100000
+
+receivers:
+  kafka:
+    brokers:
+      - localhost:9092
+    encoding: otlp_proto
+    initial_offset: earliest
+
+  otlp:
+    protocols:
+      grpc:
+      http:
+
+  jaeger:
+    protocols:
+      grpc:
+      thrift_binary:
+      thrift_compact:
+      thrift_http:
+
+  zipkin:
+
+processors:
+  batch:
+
+exporters:
+  jaeger_storage_exporter:
+    trace_storage: some_storage

cmd/jaeger/internal/integration/kafka_test.go

@@ -0,0 +1,33 @@
+// Copyright (c) 2024 The Jaeger Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package integration
+
+import (
+	"testing"
+
+	"github.com/jaegertracing/jaeger/plugin/storage/integration"
+)
+
+func TestKafkaStorage(t *testing.T) {
+	integration.SkipUnlessEnv(t, "kafka")
+
+	s := &E2EStorageIntegration{
+		ConfigFile: "../../collector-with-kafka.yaml",
+		StorageIntegration: integration.StorageIntegration{
+			CleanUp:                      purge,
+			GetDependenciesReturnsSource: true,
+			SkipArchiveTest:              true,
+		},
+	}
+
+	s.e2eInitialize(t, "kafka")
+
+	t.Cleanup(func() {
+		s.e2eCleanUp(t)
+	})
+
+	s.RunSpanStoreTests(t)
+}
+
+