Kafka batch: how to span each record

[michaldo]

Kafka batch listener takes all records polled in batch. Each record may carry own trace. When batch is done, I would like to observe it in each trace. Pseudocode:

void batchListener(List records) {
  Instant batchStart = Instant.now();
  System.out.print(records);
  for (Record record: records) {
    spanBuilder.parent(spanOf(record).startTime(batchStart).start().end();
  }
}

How to implement spanOf(Record record) ?

I see that record listener does the following

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)
...
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)

But I can't find a way to reuse existing components. As workaround I copied source code from W3CTraceContextPropagator.
I would like to see example how to do it properly

[michaldo]

Finally I solved my problem with the following code:

void batchListener(List records) {
  List<Observation> observations = records.stream().map(record ->
    KafkaListenerObservation.LISTENER_OBSERVATION.observation(
                null,
                KafkaListenerObservation.DefaultKafkaListenerObservationConvention.INSTANCE,
                () -> new KafkaRecordReceiverContext(record, "listenerId", "clientId", "groupId", () -> "cluster"),
                this.observationRegistry))
                .toList();
  observations.forEach(Observation::start);
  System.out.print(records);
  observations.forEach(Observation::end);
}

If I would like to "activate" particular observation, my code is:

Observation.Scope scope = observation.openScope();
repository.save(record.value());
scope.stop()