#5311 Fix ActiveMQ NullPointerException (#5412)

apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v4/OnExceptionInterceptor.java

@@ -34,15 +34,24 @@ import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
 public class OnExceptionInterceptor implements InstanceMethodsAroundInterceptor {
 
     public static final String CALLBACK_OPERATION_NAME_PREFIX = "RocketMQ/";
+    private static final String DEFAULT_TOPIC = "no_topic";
 
     @Override
     public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
         MethodInterceptResult result) throws Throwable {
         SendCallBackEnhanceInfo enhanceInfo = (SendCallBackEnhanceInfo) objInst.getSkyWalkingDynamicField();
-        AbstractSpan activeSpan = ContextManager.createLocalSpan(CALLBACK_OPERATION_NAME_PREFIX + enhanceInfo.getTopicId() + "/Producer/Callback");
+        String topicId = DEFAULT_TOPIC;
+        // The SendCallBackEnhanceInfo could be null when there is an internal exception in the client API,
+        // such as MQClientException("no route info of this topic")
+        if (enhanceInfo != null) {
+            topicId = enhanceInfo.getTopicId();
+        }
+        AbstractSpan activeSpan = ContextManager.createLocalSpan(CALLBACK_OPERATION_NAME_PREFIX + topicId + "/Producer/Callback");
         activeSpan.setComponent(ComponentsDefine.ROCKET_MQ_PRODUCER);
         activeSpan.errorOccurred().log((Throwable) allArguments[0]);
-        ContextManager.continued(enhanceInfo.getContextSnapshot());
+        if (enhanceInfo != null && enhanceInfo.getContextSnapshot() != null) {
+            ContextManager.continued(enhanceInfo.getContextSnapshot());
+        }
     }
 
     @Override

apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/rocketMQ/v4/OnExceptionInterceptorTest.java

@@ -65,13 +65,28 @@ public class OnExceptionInterceptorTest {
     @Before
     public void setUp() {
         exceptionInterceptor = new OnExceptionInterceptor();
+    }
 
+    @Test
+    public void testOnException() throws Throwable {
         enhanceInfo = new SendCallBackEnhanceInfo("test", contextSnapshot);
         when(enhancedInstance.getSkyWalkingDynamicField()).thenReturn(enhanceInfo);
+
+        exceptionInterceptor.beforeMethod(enhancedInstance, null, new Object[] {new RuntimeException()}, null, null);
+        exceptionInterceptor.afterMethod(enhancedInstance, null, new Object[] {new RuntimeException()}, null, null);
+
+        assertThat(segmentStorage.getTraceSegments().size(), is(1));
+        TraceSegment traceSegment = segmentStorage.getTraceSegments().get(0);
+        List<AbstractTracingSpan> spans = SegmentHelper.getSpans(traceSegment);
+        assertThat(spans.size(), is(1));
+
+        AbstractTracingSpan exceptionSpan = spans.get(0);
+        SpanAssert.assertException(SpanHelper.getLogs(exceptionSpan).get(0), RuntimeException.class);
+        SpanAssert.assertOccurException(exceptionSpan, true);
     }
 
     @Test
-    public void testOnException() throws Throwable {
+    public void testOnExceptionWithoutSkyWalkingDynamicField() throws Throwable {
         exceptionInterceptor.beforeMethod(enhancedInstance, null, new Object[] {new RuntimeException()}, null, null);
         exceptionInterceptor.afterMethod(enhancedInstance, null, new Object[] {new RuntimeException()}, null, null);
 
@@ -81,6 +96,7 @@ public class OnExceptionInterceptorTest {
         assertThat(spans.size(), is(1));
 
         AbstractTracingSpan exceptionSpan = spans.get(0);
+        assertThat(exceptionSpan.getOperationName(), is("RocketMQ/no_topic/Producer/Callback"));
         SpanAssert.assertException(SpanHelper.getLogs(exceptionSpan).get(0), RuntimeException.class);
         SpanAssert.assertOccurException(exceptionSpan, true);
     }