Fix ClassCastException from CallbackAdapterInterceptor (#6462)

CHANGES.md

@@ -16,6 +16,7 @@ Release Notes.
 * Enhance gRPC log appender to allow layout pattern.
 * Fix apm-dubbo-2.7.x-plugin memory leak due to some Dubbo RpcExceptions.
 * Fix lettuce-5.x-plugin get null host in redis sentinel mode.
+* Fix ClassCastException by making CallbackAdapterInterceptor to implement EnhancedInstance interface in the spring-kafka plugin.
 
 #### OAP-Backend
 * Allow user-defined `JAVA_OPTS` in the startup script.

apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/CallbackAdapterInterceptor.java

@@ -25,12 +25,13 @@ import org.apache.skywalking.apm.agent.core.context.ContextSnapshot;
 import org.apache.skywalking.apm.agent.core.context.tag.Tags;
 import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
 import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
 import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
 
 /**
  * implements Callback and EnhancedInstance, for kafka callback in lambda expression
  */
-public class CallbackAdapterInterceptor implements Callback {
+public class CallbackAdapterInterceptor implements Callback, EnhancedInstance {
 
     /**
      * user Callback object
@@ -64,4 +65,13 @@ public class CallbackAdapterInterceptor implements Callback {
             ContextManager.stopSpan();
         }
     }
+
+    @Override
+    public Object getSkyWalkingDynamicField() {
+        return callbackCache;
+    }
+
+    @Override
+    public void setSkyWalkingDynamicField(final Object value) {
+    }
 }
\ No newline at end of file

apm-sniffer/apm-sdk-plugin/kafka-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/CallbackInterceptorTest.java

@@ -65,24 +65,33 @@ public class CallbackInterceptorTest {
     private Object[] argumentsWithException;
     private Class[] argumentTypes;
 
-    private EnhancedInstance callBackInstance = new EnhancedInstance() {
+    private EnhancedInstance callBackInstance;
+
+    private static class CallbackInstance implements EnhancedInstance {
+        private CallbackCache cache;
+
+        public CallbackInstance(CallbackCache cache) {
+            this.cache = cache;
+        }
+
         @Override
         public Object getSkyWalkingDynamicField() {
-            CallbackCache cache = new CallbackCache();
-            cache.setSnapshot(MockContextSnapshot.INSTANCE.mockContextSnapshot());
             return cache;
         }
 
         @Override
         public void setSkyWalkingDynamicField(Object value) {
-
         }
-    };
+    }
 
     @Before
     public void setUp() {
         callbackInterceptor = new CallbackInterceptor();
 
+        CallbackCache cache = new CallbackCache();
+        cache.setSnapshot(MockContextSnapshot.INSTANCE.mockContextSnapshot());
+        callBackInstance = new CallbackInstance(cache);
+
         arguments = new Object[] {
             recordMetadata,
             null
@@ -133,6 +142,18 @@ public class CallbackInterceptorTest {
         assertCallbackSegmentRef(traceSegment.getRefs());
     }
 
+    @Test
+    public void testCallbackWithCallbackAdapterInterceptor() throws Throwable {
+        CallbackCache cacheForAdapter = new CallbackCache();
+        cacheForAdapter.setSnapshot(MockContextSnapshot.INSTANCE.mockContextSnapshot());
+        CallbackAdapterInterceptor callbackAdapterInterceptor = new CallbackAdapterInterceptor(cacheForAdapter);
+
+        CallbackCache cache = new CallbackCache();
+        cache.setCallback(callbackAdapterInterceptor);
+        EnhancedInstance instance = new CallbackInstance(cache);
+        callbackInterceptor.beforeMethod(instance, null, arguments, argumentTypes, null);
+    }
+
     private void assertCallbackSpanWithException(AbstractTracingSpan span) {
         assertCallbackSpan(span);