[test] Refactor integration tests to directly write/read ptrace.Traces

cmd/jaeger/internal/integration/trace_writer.go

@@ -28,7 +28,13 @@ var (
 	_ tracestore.Writer = (*traceWriter)(nil)
 	_ io.Closer         = (*traceWriter)(nil)
 
-	MaxChunkSize = 35 // max chunk size otel kafka export can handle safely.
+	// MaxChunkSize defines the maximum number of spans per chunk that we send via the
+	// OTLP exporter in integration tests. This was reduced from 35 to 5 when the
+	// trace writer was refactored to construct ptrace.Traces directly, so that we
+	// explicitly control chunk boundaries instead of relying on upstream batching.
+	// Smaller chunks keep the OTEL Kafka export path safely under message-size limits
+	// while still exercising the chunking logic that the Jaeger v2 pipeline depends on.
+	MaxChunkSize = 5
 )
 
 // traceWriter utilizes the OTLP exporter to send span data to the Jaeger-v2 receiver

go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/ClickHouse/clickhouse-go/v2 v2.43.0
 	github.com/apache/cassandra-gocql-driver/v2 v2.0.0
 	github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2
+	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc
 	github.com/dgraph-io/badger/v4 v4.9.1
 	github.com/elastic/go-elasticsearch/v9 v9.3.1
 	github.com/fsnotify/fsnotify v1.9.0
@@ -27,13 +28,16 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/extension/pprofextension v0.145.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/extension/sigv4authextension v0.145.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/extension/storage v0.145.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest v0.145.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil v0.145.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/jaeger v0.145.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/attributesprocessor v0.145.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/filterprocessor v0.145.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/tailsamplingprocessor v0.145.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/jaegerreceiver v0.145.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kafkareceiver v0.145.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/zipkinreceiver v0.145.0
+	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2
 	github.com/prometheus/client_golang v1.23.2
 	github.com/prometheus/client_model v0.6.2
 	github.com/prometheus/common v0.67.5
@@ -180,7 +184,6 @@ require (
 	github.com/cenkalti/backoff/v5 v5.0.3
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cpuguy83/go-md2man/v2 v2.0.6 // indirect
-	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/dgraph-io/ristretto/v2 v2.2.0 // indirect
 	github.com/dustin/go-humanize v1.0.1 // indirect
 	github.com/eapache/go-resiliency v1.7.0 // indirect
@@ -246,7 +249,6 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/kafka/configkafka v0.145.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/kafka/topic v0.145.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl v0.145.0 // indirect
-	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil v0.145.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/resourcetotelemetry v0.145.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/status v0.145.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/azure v0.145.0 // indirect
@@ -257,7 +259,6 @@ require (
 	github.com/pelletier/go-toml/v2 v2.2.4 // indirect
 	github.com/pierrec/lz4/v4 v4.1.25 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
-	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect
 	github.com/prometheus/procfs v0.19.2 // indirect
 	github.com/rcrowley/go-metrics v0.0.0-20250401214520-65e299d6c5c9 // indirect

go.sum

@@ -480,6 +480,8 @@ github.com/open-telemetry/opentelemetry-collector-contrib/pkg/batchpersignal v0.
 github.com/open-telemetry/opentelemetry-collector-contrib/pkg/batchpersignal v0.145.0/go.mod h1:3KtYwUA5QBPUQXh4fHzTS/77mHGYj0VMEuYIKnXJ0oY=
 github.com/open-telemetry/opentelemetry-collector-contrib/pkg/core/xidutils v0.145.0 h1:kkYdfEPM2bXJ39XjkactwejvyOpeJi2CSCLDLWg7p2c=
 github.com/open-telemetry/opentelemetry-collector-contrib/pkg/core/xidutils v0.145.0/go.mod h1:rMuGftjQzHzesLgLUoy7NKeZ3Hsqw9rCSJizVXX2bPs=
+github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden v0.145.0 h1:lbxy2bYh3v0YIyqd/JVttEwYlC7yU5o3JU2N/m5Qnq8=
+github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden v0.145.0/go.mod h1:kGlLjX8CJSE+9SfLARgaXTFBuAvNadjLvPsHO7fcVeE=
 github.com/open-telemetry/opentelemetry-collector-contrib/pkg/kafka/configkafka v0.145.0 h1:l/6j/BRqnKXmQfmDioatCXcWD6UYbyojKfNbbxSpD/8=
 github.com/open-telemetry/opentelemetry-collector-contrib/pkg/kafka/configkafka v0.145.0/go.mod h1:YhwSi18oBSetWlVERhz3IWl+xT5Gs/D+kuekUJyvTQc=
 github.com/open-telemetry/opentelemetry-collector-contrib/pkg/kafka/topic v0.145.0 h1:Tx55JiI9hyUwcMeGkQWCpl+1RcBYED4R2+9A6sIz00g=

internal/storage/integration/elasticsearch_test.go

@@ -17,8 +17,11 @@ import (
 	"github.com/olivere/elastic/v7"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/pdata/ptrace"
 
 	"github.com/jaegertracing/jaeger-idl/model/v1"
+	"github.com/jaegertracing/jaeger/internal/jiter"
+	"github.com/jaegertracing/jaeger/internal/jptrace"
 	escfg "github.com/jaegertracing/jaeger/internal/storage/elasticsearch/config"
 	es "github.com/jaegertracing/jaeger/internal/storage/v1/elasticsearch"
 	"github.com/jaegertracing/jaeger/internal/storage/v2/api/depstore"
@@ -250,7 +253,7 @@ func (s *ESStorageIntegration) testArchiveTrace(t *testing.T) {
 	s.skipIfNeeded(t)
 	defer s.cleanUp(t)
 	tID := model.NewTraceID(uint64(11), uint64(22))
-	expected := &model.Trace{
+	expectedV1 := &model.Trace{
 		Spans: []*model.Span{
 			{
 				OperationName: "archive_span",
@@ -262,19 +265,22 @@ func (s *ESStorageIntegration) testArchiveTrace(t *testing.T) {
 			},
 		},
 	}
+	expected := v1adapter.V1TraceToOtelTrace(expectedV1)
+	require.NoError(t, s.ArchiveTraceWriter.WriteTraces(context.Background(), expected))
 
-	require.NoError(t, s.ArchiveTraceWriter.WriteTraces(context.Background(), v1adapter.V1TraceToOtelTrace(expected)))
-
-	var actual *model.Trace
+	var actual ptrace.Traces
 	found := s.waitForCondition(t, func(_ *testing.T) bool {
-		var err error
 		iterTraces := s.ArchiveTraceReader.GetTraces(context.Background(), tracestore.GetTraceParams{TraceID: v1adapter.FromV1TraceID(tID)})
-		traces, err := v1adapter.V1TracesFromSeq2(iterTraces)
+		traces, err := jiter.CollectWithErrors(jptrace.AggregateTraces(iterTraces))
+		if err != nil {
+			t.Logf("Error loading trace: %v", err)
+			return false
+		}
 		if len(traces) == 0 {
 			return false
 		}
 		actual = traces[0]
-		return err == nil && len(actual.Spans) == 1
+		return actual.SpanCount() >= expected.SpanCount()
 	})
 	require.True(t, found)
 	CompareTraces(t, expected, actual)