[skywalking] branch master updated: Fix ClassCastException from CallbackAdapterInterceptor (#6462)

[wusheng]

This is an automated email from the ASF dual-hosted git repository.

wusheng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/skywalking.git


The following commit(s) were added to refs/heads/master by this push:
     new 3ce4dc0  Fix ClassCastException from CallbackAdapterInterceptor (#6462)
3ce4dc0 is described below

commit 3ce4dc0da8f797355bd844d22dd7da1469f342f5
Author: Jingguo Yao <yaojing...@gmail.com>
AuthorDate: Sat Feb 27 19:26:53 2021 +0800

    Fix ClassCastException from CallbackAdapterInterceptor (#6462)
---
 CHANGES.md                                         |  1 +
 .../plugin/kafka/CallbackAdapterInterceptor.java   | 12 ++++++++-
 .../apm/plugin/kafka/CallbackInterceptorTest.java  | 31 ++++++++++++++++++----
 3 files changed, 38 insertions(+), 6 deletions(-)

diff --git a/CHANGES.md b/CHANGES.md
index 0d1d554..aca21c8 100644
--- a/CHANGES.md
+++ b/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.
diff --git 
a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/CallbackAdapterInterceptor.java
 
b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/CallbackAdapterInterceptor.java
index 1717e12..835398b 100644
--- 
a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/CallbackAdapterInterceptor.java
+++ 
b/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
diff --git 
a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/CallbackInterceptorTest.java
 
b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/CallbackInterceptorTest.java
index bb6c512..78c36c2 100644
--- 
a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/CallbackInterceptorTest.java
+++ 
b/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);