From 7a3208130814d2c76fc61027a3ee4ba38a6ac466 Mon Sep 17 00:00:00 2001 From: Arunodoy18 Date: Thu, 4 Dec 2025 12:44:40 +0530 Subject: [PATCH] Fix RocketMQ context propagation for batch messages in version 5.3.4+ Context propagation was not working correctly for batch messages in RocketMQ 5.0+ because all messages in a batch were sharing the same receive span context instead of each message having its own context extracted from message properties. Changes: - Modified ReceiveSpanFinishingCallback to extract individual context for each message in the batch using MessageMapGetter to read trace headers - Each message now gets its own properly linked context for correct tracing - Re-enabled testRocketmqProduceAndBatchConsume test for latest dependencies This fixes the issue where batch message consumers would not properly propagate trace context from producers, breaking distributed tracing for RocketMQ batch message processing in versions 5.3.4+. Resolves the disabled test from #15512 --- .../rocketmqclient/v4_8/AbstractRocketMqClientTest.java | 3 --- .../rocketmqclient/v5_0/ReceiveSpanFinishingCallback.java | 8 +++++++- 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/instrumentation/rocketmq/rocketmq-client/rocketmq-client-4.8/testing/src/main/java/io/opentelemetry/instrumentation/rocketmqclient/v4_8/AbstractRocketMqClientTest.java b/instrumentation/rocketmq/rocketmq-client/rocketmq-client-4.8/testing/src/main/java/io/opentelemetry/instrumentation/rocketmqclient/v4_8/AbstractRocketMqClientTest.java index 6431b8975808..2dc0e2350012 100644 --- a/instrumentation/rocketmq/rocketmq-client/rocketmq-client-4.8/testing/src/main/java/io/opentelemetry/instrumentation/rocketmqclient/v4_8/AbstractRocketMqClientTest.java +++ b/instrumentation/rocketmq/rocketmq-client/rocketmq-client-4.8/testing/src/main/java/io/opentelemetry/instrumentation/rocketmqclient/v4_8/AbstractRocketMqClientTest.java @@ -252,9 +252,6 @@ void testRocketmqProduceAndConsume() throws Exception { @Test void testRocketmqProduceAndBatchConsume() throws Exception { - // context propagation doesn't work for batch messages in 5.3.4 - Assumptions.assumeFalse(Boolean.getBoolean("testLatestDeps")); - consumer.setConsumeMessageBatchMaxSize(2); // This test assumes that messages are sent and received as a batch. Occasionally it happens // that the messages are not received as a batch, but one by one. This doesn't match what the diff --git a/instrumentation/rocketmq/rocketmq-client/rocketmq-client-5.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/rocketmqclient/v5_0/ReceiveSpanFinishingCallback.java b/instrumentation/rocketmq/rocketmq-client/rocketmq-client-5.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/rocketmqclient/v5_0/ReceiveSpanFinishingCallback.java index 733da234b4b9..5c2c925655f9 100644 --- a/instrumentation/rocketmq/rocketmq-client/rocketmq-client-5.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/rocketmqclient/v5_0/ReceiveSpanFinishingCallback.java +++ b/instrumentation/rocketmq/rocketmq-client/rocketmq-client-5.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/rocketmqclient/v5_0/ReceiveSpanFinishingCallback.java @@ -49,8 +49,14 @@ public void onSuccess(ReceiveMessageResult receiveMessageResult) { null, timer.startTime(), timer.now()); + // For batch messages, each message should have its own context that properly + // links to the individual producer spans through context propagation for (MessageView messageView : messageViews) { - VirtualFieldStore.setContextByMessage(messageView, context); + // Extract context from individual message properties (trace headers) + Context messageContext = RocketMqSingletons.propagators() + .getTextMapPropagator() + .extract(context, messageView, MessageMapGetter.INSTANCE); + VirtualFieldStore.setContextByMessage(messageView, messageContext); } } }