Documentation Feedback on Reactor Kafka's Observation API

[almogtavor]

Documentation Issue

This PR https://github.com/reactor/reactor-kafka/pull/325/files#diff-a43be6da00bb7683292fd18443d388a44e23a6354dee611a27d07884509e101a added the first observation api & reactor kafka documentation which is great. But in general there is no Observation API & reactor (and especially no Observation API & reactor kafka) documentation, which makes the development process very confusing.
I've created this repo to test tracing with an application similar to mine: https://github.com/almogtavor/spring-boot-reactor-kafka-example. The thing is that I just can't get this to work with the observation API at all. It took me too much time to learn about the .tap operator, and regarding the KafkaReceiverObservation & similar classes, I still don't understand how they shall be used. Is there an example of the ideal usage of the observation API for tracing in a reactor kafka-based application that reads data from one topic and sends it to another?

Improvement Suggestion

The documentation that the new PR will add is awesome, but I it's essential to provide specific guidelines addressing common scenarios:

1. Parallel Publishers: How to add traces to an application that uses publishers like .publishOn(Schedulers.parallel()). This publisher didn't work well with the automatic instrumentation that Otel did with the javaagent, therefore I started to migrate to the manual instrumentation.
2. Buffer Operator: For an application that uses the .buffer operator. In my case I want the traces of the batch operations to have links from and to every original trace ID of the messages that were included in the buffer. (If it's relevant - my use case is buffering 50 messages and performing a Mongo write after buffering. I want to be able to create 51 spans - 1 for the whole buffer and 50 for each message, and then link the buffer's span to the messages').
3. Context Management: Post my attempt to transition to the native Otel API, I found it challenging to utilize the observation context for tasks like adding the Span object Understanding reactor + reactor-kafka instrumentations open-telemetry/opentelemetry-java-instrumentation#9771. A simple example of a reactor kafka application that puts a key and gets it at the end of the flux could help with this.

Additional context

While the ongoing efforts to improve documentation are appreciated, there's a need for clearer, more structured, and example-driven documentation to facilitate a smoother development experience.

[janchristian-haddorp]

@almogtavor, first you can try replacing ReactiveKafkaProducerTemplate by KafkaSender. This allows to confiure metrics and observation. Will be available on upcoming Reactor Kafka release.

@Bean
public SenderOptions<String, OutputPojo> kafkaProducerOptions(
        @Value("${spring.kafka.consumer.topic}") String outputTopic,
        KafkaProperties properties) {
    Map<String, Object> props = properties.buildProducerProperties();

    return SenderOptions.create(props);
}

@Bean
public KafkaSender<String, OutputPojo> kafkaSender(
        SenderOptions<String, OutputPojo> kafkaSenderOptions,
        ObservationRegistry registry) {
    return KafkaSender.create(kafkaSenderOptions
            .withObservation(registry));
}

Beside this using libs on your sample project look slightly outdated. Not sure if this still fits together...

[patpatpat123]

upvoting this

[almogtavor]

@janchristian-haddorp thank you very much. I didn't know about the .withObservation method so I'll try it and update my deps to boot 3 etc. Can you refer to my 3 points I struggled with? Have you tried tracing the parallel scheduler? Buffers?

[artembilan]

See an Observation support in the upcoming release: https://github.com/reactor/reactor-kafka/blob/main/src/docs/asciidoc/api-guide.adoc#micrometer-observation.

The rest of questions about parallel scheduler and buffers are out of this project scope.
See general Reactor docs:
https://projectreactor.io/docs/core/release/reference/#_observation
https://projectreactor.io/docs/core/release/reference/#context.propagation

The Otel API is definitely out of our reach at all.

Not sure what else you would like to see from the scope of this project...

[michalwilk96]

@artembilan what is the usage of "user.receiver" inside KafkaRecordReceiverContext? (https://github.com/reactor/reactor-kafka/blob/main/src/docs/asciidoc/api-guide.adoc#micrometer-observation)

[artembilan]

See KafkaRecordReceiverContext source code: https://github.com/reactor/reactor-kafka/blob/main/src/main/java/reactor/kafka/receiver/observation/KafkaRecordReceiverContext.java#L37

So, that is a receiverId - a unique id for the reactor.kafka.client.id tag in the target span.
Logical identifier of your Reactor Kafka service.