[v2] Add legacy formats into e2e kafka integration tests

cmd/jaeger/collector-with-jaeger.yaml

@@ -0,0 +1,33 @@
+service:
+  pipelines:
+    traces:
+      receivers: [otlp, jaeger]
+      processors: [batch]
+      exporters: [kafka]
+  telemetry:
+    resource:
+      service.name: jaeger_collector
+    metrics:
+      level: detailed
+
+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: jaeger_json

cmd/jaeger/ingester-with-jaeger.yaml

@@ -0,0 +1,37 @@
+service:
+  extensions: [jaeger_storage, jaeger_query]
+  pipelines:
+    traces:
+      receivers: [kafka]
+      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
+    topic: "jaeger-spans"
+    encoding: jaeger_json
+    initial_offset: earliest
+
+processors:
+  batch:
+
+exporters:
+  jaeger_storage_exporter:
+    trace_storage: some_storage

cmd/jaeger/internal/integration/kafka_test.go

@@ -20,30 +20,65 @@ func TestKafkaStorage(t *testing.T) {
 	// we can change the config to use topic: "${KAFKA_TOPIC:-jaeger-spans}"
 	// and export a KAFKA_TOPIC var with random topic name in the tests.
 
-	collectorConfig := "../../collector-with-kafka.yaml"
-	ingesterConfig := "../../ingester-remote-storage.yaml"
-
-	collector := &E2EStorageIntegration{
-		SkipStorageCleaner:  true,
-		ConfigFile:          collectorConfig,
-		HealthCheckEndpoint: "http://localhost:8888/metrics",
-	}
-
-	// Initialize and start the collector
-	collector.e2eInitialize(t, "kafka")
-
-	ingester := &E2EStorageIntegration{
-		ConfigFile: ingesterConfig,
-		StorageIntegration: integration.StorageIntegration{
-			CleanUp:                      purge,
-			GetDependenciesReturnsSource: true,
-			SkipArchiveTest:              true,
-		},
-	}
-
-	// Initialize and start the ingester
-	ingester.e2eInitialize(t, "kafka")
-
-	// Run the span store tests
-	ingester.RunSpanStoreTests(t)
+	// OTLP config files
+	otlpCollectorConfig := "../../collector-with-kafka.yaml"
+	otlpIngesterConfig := "../../ingester-remote-storage.yaml"
+
+	// Legacy Jaeger config files
+	legacyCollectorConfig := "../../collector-with-jaeger.yaml"
+	legacyIngesterConfig := "../../ingester-with-jaeger.yaml"
+
+	// Test OTLP formats
+	t.Run("OTLPFormats", func(t *testing.T) {
+		collector := &E2EStorageIntegration{
+			SkipStorageCleaner:  true,
+			ConfigFile:          otlpCollectorConfig,
+			HealthCheckEndpoint: "http://localhost:8888/metrics",
+		}
+
+		// Initialize and start the OTLP collector
+		collector.e2eInitialize(t, "kafka")
+
+		ingester := &E2EStorageIntegration{
+			ConfigFile: otlpIngesterConfig,
+			StorageIntegration: integration.StorageIntegration{
+				CleanUp:                      purge,
+				GetDependenciesReturnsSource: true,
+				SkipArchiveTest:              true,
+			},
+		}
+
+		// Initialize and start the OTLP ingester
+		ingester.e2eInitialize(t, "kafka")
+
+		// Run the span store tests for OTLP formats
+		ingester.RunSpanStoreTests(t)
+	})
+
+	// Test legacy Jaeger formats
+	t.Run("LegacyJaegerFormats", func(t *testing.T) {
+		collector := &E2EStorageIntegration{
+			SkipStorageCleaner:  true,
+			ConfigFile:          legacyCollectorConfig,
+			HealthCheckEndpoint: "http://localhost:8888/metrics",
+		}
+
+		// Initialize and start the Jaeger collector
+		collector.e2eInitialize(t, "kafka")
+
+		ingester := &E2EStorageIntegration{
+			ConfigFile: legacyIngesterConfig,
+			StorageIntegration: integration.StorageIntegration{
+				CleanUp:                      purge,
+				GetDependenciesReturnsSource: true,
+				SkipArchiveTest:              true,
+			},
+		}
+
+		// Initialize and start the Jaeger ingester
+		ingester.e2eInitialize(t, "kafka")
+
+		// Run the span store tests for legacy Jaeger formats
+		ingester.RunSpanStoreTests(t)
+	})
 }