diff --git a/CHANGES.md b/CHANGES.md index 0d1d55452..aca21c892 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 1717e12e3..835398b16 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 bb6c512a0..78c36c2b7 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);