[Java Agent] NPE in rocketMQ-5.x-plugin OnSuccessInterceptor when rocketmq-client 5.x wraps async-send callbacks (BackpressureSendCallBack)

[MichaelShawn-X]

Search before asking

• I had searched in the issues and found no similar issues.

Apache SkyWalking agent version

9.8.0 (latest release). The affected code is byte-for-byte identical on main (diffed OnSuccessInterceptor.java, SendCallbackInstrumentation.java, MessageSendInterceptor.java against the v9.8.0 tag — no difference), so no released or in-development version fixes this yet.

Target / library versions

• org.apache.rocketmq:rocketmq-client:5.3.1 — the Remoting-protocol legacy client (not the rocketmq-client-java gRPC SDK)
• Deployed through the Spring Cloud Stream RocketMQ binder (Spring Cloud Alibaba 2025.0.0.0), which enables client message trace by default, so both the business-callback chain and the AsyncTraceDispatcher chain are exercised
• JDK 21, Linux containers (K8s)

What happened

Under load (high-frequency async sends), the agent log continuously shows two paired ERRORs on the RocketMQ client callback threads:

ERROR InstMethodsInter : class[class org.apache.rocketmq.client.trace.AsyncTraceDispatcher$AsyncDataSendTask$1] before method[onSuccess] intercept failure
java.lang.NullPointerException
ERROR InstMethodsInter : class[class org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl$BackpressureSendCallBack] after method[onSuccess] intercept failure
java.lang.NullPointerException

The exceptions are swallowed by InstMethodsInter (host methods keep running), but the callback spans are recorded incorrectly and the log noise is constant at high throughput.

Root cause

Three facts combine into a guaranteed NPE whenever one send produces a nested chain of two SendCallback implementations:

1. The plugin enhances every implementation of SendCallback via hierarchy match:
   https://github.com/apache/skywalking-java/blob/v9.8.0/apm-sniffer/apm-sdk-plugin/rocketMQ-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v5/define/SendCallbackInstrumentation.java#L84

2. rocketmq-client 5.x wraps every async-send callback one level deeper. Decompiling 5.3.1 confirms DefaultMQProducerImpl holds a private final SendCallback sendCallback delegate inside a new internal wrapper and constructs it on the send path:
   new BackpressureSendCallBack(this, callback). The trace dispatcher's AsyncTraceDispatcher$AsyncDataSendTask$1 is the delegate in the msgTrace chain.

3. MessageSendInterceptor injects SendCallBackEnhanceInfo (topic + context snapshot) only into the outermost callback object it sees at MQClientAPIImpl.sendMessage — i.e. the BackpressureSendCallBack wrapper. The inner delegate never receives it:
   https://github.com/apache/skywalking-java/blob/v9.8.0/apm-sniffer/apm-sdk-plugin/rocketMQ-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v5/MessageSendInterceptor.java#L97-L100

Then, on callback execution:

4. OnSuccessInterceptor.beforeMethod dereferences the dynamic field without a null check, so the inner delegate hits NPE at enhanceInfo.getTopicId():
   https://github.com/apache/skywalking-java/blob/v9.8.0/apm-sniffer/apm-sdk-plugin/rocketMQ-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v5/OnSuccessInterceptor.java#L45-L46

5. afterMethod calls ContextManager.stopSpan() unconditionally:
   https://github.com/apache/skywalking-java/blob/v9.8.0/apm-sniffer/apm-sdk-plugin/rocketMQ-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v5/OnSuccessInterceptor.java#L59

The observed log sequence maps exactly onto this nesting (outer callback runs first, delegates to inner in the same call stack):

Step	Event	Result
1	Outer BackpressureSendCallBack.beforeMethod	OK — dynamic field present, local span created
2	Delegates to inner AsyncDataSendTask$1.beforeMethod	NPE — dynamic field is null → before ... intercept failure (1st ERROR)
3	Inner afterMethod	stopSpan() pops the outer span it did not create
4	Outer afterMethod	stopSpan() on an empty span stack → NPE → after ... intercept failure (2nd ERROR)

Note the log signature fits only this explanation: the inner callback reports a before failure, the outer reports an after failure, and there is no outer before error — any other failure mode would produce a different pair.

Impact

• Message delivery, msgTrace reporting and business callbacks are unaffected (the agent swallows the exception), so this is not a data-path bug.
• The outer callback's local span gets popped/stopped by the inner afterMethod, so callback spans are misattributed or dropped and traces break; at throughput the paired ERRORs are constant log noise.

Reproduce

Attach agent 9.8.0 to any service using rocketmq-client 5.3.x with at least one async send, e.g.:

DefaultMQProducer producer = new DefaultMQProducer("demo");
producer.start();
producer.send(msg, new SendCallback() {          // wrapped internally by BackpressureSendCallBack
    public void onSuccess(SendResult r) { }
    public void onException(Throwable t) { }
}, 3000);

With msgTrace enabled (as with the Spring Cloud Stream RocketMQ binder defaults), the AsyncTraceDispatcher chain triggers the exact same pattern on every trace message.

Suggested fix

• In OnSuccessInterceptor.beforeMethod, bail out when objInst.getSkyWalkingDynamicField() is null (debug-log and return without creating a span or calling continued).
• Pair afterMethod with "did beforeMethod actually create a span" — e.g. a thread-local marker set only on successful span creation — instead of the unconditional stopSpan(), so a failed nested callback cannot pop its parent's span.
• Optionally narrow the hierarchy match to exclude client-internal wrapper callbacks, though the two changes above are the robust fix (the wrappers will keep evolving across client versions).

Additional context

The 4.x plugin had a sibling problem with SendCallback interception before (#5311). This issue is the same class of bug surfacing in the 5.x plugin against client 5.x's internal callback wrappers. Happy to provide further decompiled details of rocketmq-client:5.3.1 if useful.

[wu-sheng]

Thanks for the detailed report and analysis. Confirmed, and fixed in apache/skywalking-java#848, which will ship in 9.9.0.

• Since rocketmq-client 5.1.4, all three async send(..., SendCallback, ...) overloads wrap the callback in BackpressureSendCallBack, whether backpressure is enabled or not. So any async send with a callback is affected, not only msgTrace.
• Only the wrapper reaches MQClientAPIImpl#sendMessage and gets the SendCallBackEnhanceInfo. Your sequence is right: the inner callback's beforeMethod throws the NPE, its unconditional stopSpan() pops the wrapper's span, and the wrapper's afterMethod throws the second NPE on the empty context. onException already handles the missing info.
• The fix: OnSuccessInterceptor skips beforeMethod, afterMethod and handleMethodException when the callback has no info. Nothing sets that field during the call, so the three stay paired without a thread-local marker. The wrapper keeps the span and the context continuation, and your callback runs inside that span.

On impact: when the callback thread has no other active span, which is the usual case, the wrapper's span is still finished and reported with the right parent, so the trace data stays intact. What you see is the two ERRORs per callback. A trace would only be corrupted if the callback thread already had an active span.

While checking newer RocketMQ versions, the same PR also fixes two issues in the rocketMQ-client-java-5.x plugin (the gRPC SDK), and adds rocketmq-client 5.5 and rocketmq-client-java 5.0, 5.1 and 5.2 to the plugin tests.