ProducerMessageHandlerCustomizer doesn't allow to configure the Micrometer Tracing Observation

[codependent]

The issue is thouroughly described in the following Stackoverflow question: https://stackoverflow.com/questions/74666562/how-to-instrument-spring-boot-3-x-with-spring-cloud-stream-4-x-producers-and-con --> UPDATES 1 & 2

Version of the framework Spring Boot 3 - Spring Cloud Stream 4.0.0-RC3

[sobychacko]

Hi, @codependent, Thanks for reporting the issue and the context on the SO thread. Any chance you can create a simple application where we can reproduce the issue and try some fixes?

[codependent]

Hi, @codependent, Thanks for reporting the issue and the context on the SO thread. Any chance you can create a simple application where we can reproduce the issue and try some fixes?

Sure!

[codependent]

I've uploaded a sample project reproducing the problem: https://github.com/codependent/kotlin-kafka-correlation

It just needs a local Kafka cluster on the default port.

After starting the application just invoke curl localhost:8080/publish

Relevant code:

The controller is wrapped with withObservationContext(observationRegistry) { to setup the coroutine context with the tracing information used in MDC:

        withObservationContext(observationRegistry) {
            eventProducer.send(RandomStringUtils.randomAlphanumeric(10), RandomStringUtils.randomAlphanumeric(10))
        }

In order to propagate the tracing information in the Kafka Producer I had to define the following ProducerMessageHandlerCustomizer as indicated by Gary Russell.

@Configuration
class KafkaProducerConfiguration {

    @Bean
    fun kafkaProducerObservationCustomizer (applicationContext: ApplicationContext, observationRegistry: ObservationRegistry) : ProducerMessageHandlerCustomizer<KafkaProducerMessageHandler<String, String>> {
        return ProducerMessageHandlerCustomizer<KafkaProducerMessageHandler<String, String>> {
                handler, destinationName ->
            handler.kafkaTemplate.setObservationEnabled(true)
            handler.kafkaTemplate.setApplicationContext(applicationContext)
            handler.kafkaTemplate.afterSingletonsInstantiated()
        }
    }
}

This additional calls:

            handler.kafkaTemplate.setApplicationContext(applicationContext)
            handler.kafkaTemplate.afterSingletonsInstantiated()

...are a workaround for this issue, as described in the SO question (UPDATE 1 & 2)

With it in place we can now see that the traceparent header is added to the message:

2022-12-07T15:31:47.134+01:00  INFO [,6390a3d35cd98bf3bf2ac41a0fc81c2f,bf2ac41a0fc81c2f] 18586 --- [ctor-http-nio-3] c.c.c.kafka.producer.EventProducer       : Sending message to kafka queue GenericMessage [payload=WSuP1KBpyT, headers={kafka_timestamp=1670423507134, id=79dcb7b8-1d12-3de4-4e11-9f6802212700, kafka_messageKey=42QIJ3H7Kd, timestamp=1670423507134}]

and received in the consumers:

2022-12-07T15:31:47.136+01:00  INFO [,,] 18586 --- [-StreamThread-1] ventKStreamConfiguration$$SpringCGLIB$$0 : processEvents() -> 
headers [
   contentType - application/json, 
   target-protocol - kafka, 
   spring_json_header_types - {"contentType":"java.lang.String","target-protocol":"java.lang.String"}, 
   traceparent - 00-6390a3d35cd98bf3bf2ac41a0fc81c2f-c52c9346b1e88729-00]  
key 42QIJ3H7Kd - 
value WSuP1KBpyT
2022-12-07T15:31:47.137+01:00  INFO [,,] 18586 --- [container-0-C-1] fkaConsumerConfiguration$$SpringCGLIB$$0 : consumeEvents() -> 
headers [
   deliveryAttempt - 1, 
   kafka_timestampType - CREATE_TIME, 
   kafka_receivedMessageKey - [B@293d0088, 
   kafka_receivedTopic - events, 
   kafka_offset - 21, 
   scst_nativeHeadersPresent - true, 
   traceparent - [B@38690aa7, 
   kafka_consumer - org.apache.kafka.clients.consumer.KafkaConsumer@63ae5309, 
   source-type - kafka, 
   id - 99be87fd-d0a5-f9d9-2468-2b2dd38d17b3, 
   kafka_receivedPartitionId - 0, 
   contentType - application/json, 
   kafka_receivedTimestamp - 1670423507134, 
   kafka_groupId - anonymous.721f9969-21bd-4810-9470-15c7375617c4, 
   timestamp - 1670423507137]
payload WSuP1KBpyT

The consumers logs (a regular Consumer and a KStream) aren't automatically correlating the traceId/spainId in the logs (INFO [,,] ). There's a specific issue for this problem: #2577

[codependent]

@sobychacko @garyrussell
I'm experiencing a problem that was already indicated in the SO question, when the Kafka cluster is remote:

Working with the latest snapshot I've configured:

spring:
  cloud:
    stream:
      kafka:
        binder:
          enableObservation: true

When the first message is going to be sent and the producer binding is started, KafkaTemplate.afterSingletonsInstantiated() is invoked, accessing the clusterId.

this.clusterId = this.kafkaAdmin.clusterId();

When KafkaAdmin.createdAdmin() is called, it only has one configs entry: bootstrap.servers -> 9092.
KafkaAdmin properties come from KafkaProperties (prefix spring.kafka) instead of the spring.cloud.stream.kafka.binder.properties. It seems that when enableObservation is true I should also configure spring.kafka properties, which is kind of redundant and isn't necessary when enableObservation is false.

Without that config there's an error when it tries to connect to a local cluster instead of the actual remote cluster:

[AdminClient clientId=adminclient-2] Node -1 disconnected.
[AdminClient clientId=adminclient-2] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.

What do you think?

[garyrussell]

The problem is that the binder's template is finding Boot's auto configured RabbitAdmin; we will need to think how to solve that; it will probably need some changes in the template itself (as well as the binder). Please open an issue against spring-kafka referencing this comment. Thanks.

In the meantime, your work around is correct - it will cause Boot's auto configured admin to get the cluster id from your actual cluster.

The cluster id is needed to identify the cluster in the traces.

[garyrussell]

By the way, you don't have to set the bootstrap servers in both places; the binder will use the spring.kafka.* properties first, before looking for binder-specific properties.