traceId is not being propagated from RabbitListener through its return

[TJackler]

Hi,

I am designing a system that takes a REST call and sends a message to rabbit a queue. The RabbitListener method for that queue processes the message with its handler, but when I take the message and pass to another queue, the traceId is not propagated to the RabbitListener handler of the new queue.

For visualization purpose;

User -> Rest -> queue1 -> RabbitListner(queue1) (correct traceId) -> queue2 -> RabbitListener(queue2) (wrong traceId)t -> Back to Rest Call

@Component
public class MessageHandler {

    private static final Logger log = LoggerFactory.getLogger(MessageHandler.class);

    private final Executor executor = Executors.newCachedThreadPool();

    @RabbitListener(queuesToDeclare = @Queue(name = "task-queue", durable = "durable", autoDelete = "true"), ackMode = "MANUAL")
    public CompletableFuture<String> handleMessage(String s, @Headers Map<String, Object> headers) {
        log.info("message received: {}", s);
        return CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(1000);
                log.info("sending reply: {}", s.toUpperCase());
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return s.toUpperCase();
        }, executor);
    }
}

The header (which comes from the rest through rabbitTemplate) includes traceparent but. Returned message (from this method) to desired queue does not include traceparent header.

I checked out the examples below; but It did not help to propagate traceId for this case.
https://github.com/micrometer-metrics/micrometer-samples/tree/main/rabbitmq-consumer
https://github.com/micrometer-metrics/micrometer-samples/tree/main/rabbitmq-producer

Thanks for the development @marcingrzejszczak and rest of the team members.

[marcingrzejszczak]

Are you saying that our samples do not work well?

[github-actions]

If you would like us to look at this issue, please provide the requested information. If the information is not provided within the next 7 days this issue will be closed.

[TJackler]

Your examples work fine for the case which is not changing the thread. When I use CompletableFuture.supplyAsync and supply the job taken from the queue to the another thread (due to supplyAsync), it does not propagate the trace id to that thread.

[marcingrzejszczak]

have you passed an executor to the supplyAsync and wrapped it with context propagation?

[github-actions]

If you would like us to look at this issue, please provide the requested information. If the information is not provided within the next 7 days this issue will be closed.

[TJackler]

Yes, I did wrap the custom context propagator to the supply async, but it does not make traceId propagation possible to the new thread local.

[marcingrzejszczak]

Can you file an issue in Spring AMQP project where the actual instrumentation takes place?

[github-actions]

If you would like us to look at this issue, please provide the requested information. If the information is not provided within the next 7 days this issue will be closed.

[marcingrzejszczak]

Closing in favour of spring-projects/spring-amqp#2608