Fix issue that webfluxwebclient plugin async finish repeatedly in multi thread (#7229)

CHANGES.md

@@ -26,6 +26,7 @@ Release Notes.
 * Support parameter collection for SqlServer.
 * Add `ShardingSphere-5.0.0-beta` plugin.
 * Fix some method exception error.
+* Fix async finish repeatedly in `spring-webflux-5.x-webclient` plugin.
 
 #### OAP-Backend
 

apm-sniffer/apm-sdk-plugin/spring-plugins/spring-webflux-5.x-webclient-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/webflux/v5/webclient/WebFluxWebClientInterceptor.java

@@ -24,8 +24,8 @@ 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.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
-import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.v2.InstanceMethodsAroundInterceptorV2;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.v2.MethodInvocationContext;
 import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
 import org.springframework.http.HttpStatus;
 import org.springframework.web.reactive.function.client.ClientRequest;
@@ -36,10 +36,10 @@ import java.lang.reflect.Method;
 import java.net.URI;
 import java.util.function.BiConsumer;
 
-public class WebFluxWebClientInterceptor implements InstanceMethodsAroundInterceptor {
+public class WebFluxWebClientInterceptor implements InstanceMethodsAroundInterceptorV2 {
 
     @Override
-    public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
+    public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes, MethodInvocationContext context) throws Throwable {
         if (allArguments[0] == null) {
             //illegal args,can't trace ignore
             return;
@@ -63,22 +63,22 @@ public class WebFluxWebClientInterceptor implements InstanceMethodsAroundInterce
         if (request instanceof EnhancedInstance) {
             ((EnhancedInstance) request).setSkyWalkingDynamicField(contextCarrier);
         }
-        
+
         //user async interface
         span.prepareForAsync();
         ContextManager.stopSpan();
 
-        objInst.setSkyWalkingDynamicField(span);
+        context.setContext(span);
     }
 
     @Override
-    public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes, Object ret) throws Throwable {
+    public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes, Object ret, MethodInvocationContext context) throws Throwable {
         // fix the problem that allArgument[0] may be null
         if (allArguments[0] == null) {
             return ret;
         }
         Mono<ClientResponse> ret1 = (Mono<ClientResponse>) ret;
-        AbstractSpan span = (AbstractSpan) objInst.getSkyWalkingDynamicField();
+        AbstractSpan span = (AbstractSpan) context.getContext();
         return ret1.doAfterSuccessOrError(new BiConsumer<ClientResponse, Throwable>() {
             @Override
             public void accept(ClientResponse clientResponse, Throwable throwable) {
@@ -98,7 +98,7 @@ public class WebFluxWebClientInterceptor implements InstanceMethodsAroundInterce
     }
 
     @Override
-    public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes, Throwable t) {
+    public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes, Throwable t, MethodInvocationContext context) {
         AbstractSpan activeSpan = ContextManager.activeSpan();
         activeSpan.errorOccurred();
         activeSpan.log(t);

apm-sniffer/apm-sdk-plugin/spring-plugins/spring-webflux-5.x-webclient-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/webflux/v5/webclient/define/WebFluxWebClientInstrumentation.java

@@ -21,15 +21,15 @@ package org.apache.skywalking.apm.plugin.spring.webflux.v5.webclient.define;
 import net.bytebuddy.description.method.MethodDescription;
 import net.bytebuddy.matcher.ElementMatcher;
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
-import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.v2.InstanceMethodsInterceptV2Point;
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.StaticMethodsInterceptPoint;
-import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassEnhancePluginDefine;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.v2.ClassInstanceMethodsEnhancePluginDefineV2;
 import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
 import org.apache.skywalking.apm.agent.core.plugin.match.NameMatch;
 
 import static net.bytebuddy.matcher.ElementMatchers.named;
 
-public class WebFluxWebClientInstrumentation extends ClassEnhancePluginDefine {
+public class WebFluxWebClientInstrumentation extends ClassInstanceMethodsEnhancePluginDefineV2 {
     private static final String ENHANCE_CLASS = "org.springframework.web.reactive.function.client.ExchangeFunctions$DefaultExchangeFunction";
     private static final String INTERCEPT_CLASS = "org.apache.skywalking.apm.plugin.spring.webflux.v5.webclient.WebFluxWebClientInterceptor";
 
@@ -44,16 +44,16 @@ public class WebFluxWebClientInstrumentation extends ClassEnhancePluginDefine {
     }
 
     @Override
-    public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
-        return new InstanceMethodsInterceptPoint[]{
-                new InstanceMethodsInterceptPoint() {
+    public InstanceMethodsInterceptV2Point[] getInstanceMethodsInterceptV2Points() {
+        return new InstanceMethodsInterceptV2Point[]{
+                new InstanceMethodsInterceptV2Point() {
                     @Override
                     public ElementMatcher<MethodDescription> getMethodsMatcher() {
                         return named("exchange");
                     }
 
                     @Override
-                    public String getMethodsInterceptor() {
+                    public String getMethodsInterceptorV2() {
                         return INTERCEPT_CLASS;
                     }