Tracing with Kafka batch listener

[michaldo]

While Kafka tracing is quite simple for Kafka record listener, I have unexpected problems with batch listener.
My goal: when batch listener receives batch of 2 records, each may have different trace id. When batch processing is done, I would like to "close" 2 spans. How to do it properly?

https://spring.io/blog/2022/10/12/observability-with-spring-boot-3 shows how to close one "global" span with Observation API. It is not my case, then I used "Span API".
My biggest problem was extracting trace from each record. Existing code seems be hard to use from outside.
Finally, I copied source code from W3CTraceContextPropagator to own class to use W3CTraceContextPropagator#extractContextFromTraceParent

I would like to see example how to manage traces with batch listener

[jkwatson]

Since this is specifically about instrumentation, it might be more appropriate to ask over in the instrumentation repository. (https://github.com/open-telemetry/opentelemetry-java-instrumentation)

[michaldo]

To clarify what I need, I copied stacktrace how single record listener extracts span from Kafka message.
I would like to do the same, for batch listener, without copy source code from W3CTraceContextPropagator

extractContextFromTraceParent:202, W3CTraceContextPropagator (io.opentelemetry.api.trace.propagation)
extractImpl:155, W3CTraceContextPropagator (io.opentelemetry.api.trace.propagation)
extract:141, W3CTraceContextPropagator (io.opentelemetry.api.trace.propagation)
extract:65, MultiTextMapPropagator (io.opentelemetry.context.propagation)
lambda$extract$2:110, CompositeTextMapPropagator (org.springframework.boot.actuate.autoconfigure.tracing)
apply:-1, CompositeTextMapPropagator$$Lambda/0x0000018c01aa4c80 (org.springframework.boot.actuate.autoconfigure.tracing)
accept:197, ReferencePipeline$3$1 (java.util.stream)
tryAdvance:708, AbstractList$RandomAccessSpliterator (java.util)
forEachWithCancel:129, ReferencePipeline (java.util.stream)
copyIntoWithCancel:527, AbstractPipeline (java.util.stream)
copyInto:513, AbstractPipeline (java.util.stream)
wrapAndCopyInto:499, AbstractPipeline (java.util.stream)
evaluateSequential:150, FindOps$FindOp (java.util.stream)
evaluate:234, AbstractPipeline (java.util.stream)
findFirst:647, ReferencePipeline (java.util.stream)
extract:112, CompositeTextMapPropagator (org.springframework.boot.actuate.autoconfigure.tracing)
extract:65, OtelPropagator (io.micrometer.tracing.otel.bridge)
onStart:58, PropagatingReceiverTracingObservationHandler (io.micrometer.tracing.handler)
onStart:35, PropagatingReceiverTracingObservationHandler (io.micrometer.tracing.handler)
onStart:149, ObservationHandler$FirstMatchingCompositeObservationHandler (io.micrometer.observation)
notifyOnObservationStarted:214, SimpleObservation (io.micrometer.observation)
start:161, SimpleObservation (io.micrometer.observation)
observe:563, Observation (io.micrometer.observation)
doInvokeRecordListener:2699, KafkaMessageListenerContainer$ListenerConsumer (org.springframework.kafka.listener)

[jkwatson]

yes, I suggest asking about this in the instrumentation repository. thanks!

[michaldo]

I asked open-telemetry/opentelemetry-java-instrumentation#12276

Finally I replaced Span API by observation API and more or less solved my problem. Details in linked discussion