How to get correlation id in Spring Cloud Stream

May 10, 2023

Spring Team,

The below producer is able to send the values to a kafka topic successfully.

@Bean
Supplier<Flux<Integer>> someProducer(){
    return () -> Flux.range(1, 10);
}

But..how do we get the correlation id of the message produced as we get using ReactiveKafkaSender? Since the flux is subscribed by Spring internally, Is there any way to get?

>Solution :

Currently, the binder does not support getting the complete SenderResult, only the RecordMetadata of a successful send.

Please open a bug on GitHub (spring-cloud-stream) and reference this question.

To get the RecordMetadata you can use Supplier<Flux<Message<Integer>>> and set the senderResult header in the message to an AtomicInteger<Mono<RecordMetadata>>; it will be populated with a Mono which you can subscribe to.

There’s a test here: https://github.com/spring-cloud/spring-cloud-stream/blob/29c3cd7cddf9b853c57fca2b2118f1b64e5dde30/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderTests.java#L315-L323

However, I can see that this is not much use without the correlation metadata.