Fix the WebClient plugins' NPE when writeTo runs before the exit spanis created

CHANGES.md

@@ -5,6 +5,12 @@ Release Notes.
 9.8.0
 ------------------
 
+* Fix the `NullPointerException` thrown by the `spring-webflux-5.x-webclient` and
+  `spring-webflux-6.x-webclient` plugins when `DefaultClientRequestBuilder$BodyInserterRequest#writeTo` runs
+  before any exit span exists. Connectors such as `JdkClientHttpConnector` call `writeTo` eagerly at assembly
+  time, while the exchange interceptor creates the exit span and its `ContextCarrier` only at subscription, so
+  the interception failed and the `sw8` header was not propagated. The carrier injection is now null-guarded
+  and, if the carrier is still absent, retried when the returned `Mono` is subscribed (apache/skywalking#13589).
 * Fix the `spring-ai-1.x-plugin` `ChatModelStreamInterceptor` leaking its async span when
   `ChatModel#stream(Prompt)` fails synchronously, which silently dropped the whole `TraceSegment`
   of the request (apache/skywalking#14098).

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/BodyInserterRequestInterceptor.java

@@ -24,6 +24,7 @@
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
 import org.springframework.http.client.reactive.ClientHttpRequest;
+import reactor.core.publisher.Mono;
 
 import java.lang.reflect.Method;
 
@@ -34,17 +35,40 @@ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allAr
                              MethodInterceptResult result) throws Throwable {
         ClientHttpRequest clientHttpRequest = (ClientHttpRequest) allArguments[0];
         ContextCarrier contextCarrier = (ContextCarrier) objInst.getSkyWalkingDynamicField();
-        CarrierItem next = contextCarrier.items();
-        while (next.hasNext()) {
-            next = next.next();
-            clientHttpRequest.getHeaders().set(next.getHeadKey(), next.getHeadValue());
+        if (contextCarrier != null) {
+            inject(clientHttpRequest, contextCarrier);
         }
     }
 
     @Override
     public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
                               Object ret) throws Throwable {
-        return ret;
+        // Connectors like JdkClientHttpConnector invoke writeTo eagerly at assembly time,
+        // before the exchange interceptor sets the carrier at subscription time. Retry the
+        // injection when the returned Mono is subscribed, before the request is committed.
+        if (objInst.getSkyWalkingDynamicField() != null || !(ret instanceof Mono)) {
+            return ret;
+        }
+        final ClientHttpRequest clientHttpRequest = (ClientHttpRequest) allArguments[0];
+        return Mono.defer(() -> {
+            ContextCarrier contextCarrier = (ContextCarrier) objInst.getSkyWalkingDynamicField();
+            if (contextCarrier != null) {
+                try {
+                    inject(clientHttpRequest, contextCarrier);
+                } catch (Throwable t) {
+                    // headers are read-only once the request is committed (e.g. re-subscribed by a retry)
+                }
+            }
+            return (Mono<?>) ret;
+        });
+    }
+
+    private void inject(ClientHttpRequest clientHttpRequest, ContextCarrier contextCarrier) {
+        CarrierItem next = contextCarrier.items();
+        while (next.hasNext()) {
+            next = next.next();
+            clientHttpRequest.getHeaders().set(next.getHeadKey(), next.getHeadValue());
+        }
     }
 
     @Override

apm-sniffer/apm-sdk-plugin/spring-plugins/spring-webflux-6.x-webclient-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/webflux/v6/webclient/BodyInserterRequestInterceptor.java

@@ -24,6 +24,7 @@
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
 import org.springframework.http.client.reactive.ClientHttpRequest;
+import reactor.core.publisher.Mono;
 
 import java.lang.reflect.Method;
 
@@ -34,17 +35,40 @@ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allAr
                              MethodInterceptResult result) throws Throwable {
         ClientHttpRequest clientHttpRequest = (ClientHttpRequest) allArguments[0];
         ContextCarrier contextCarrier = (ContextCarrier) objInst.getSkyWalkingDynamicField();
-        CarrierItem next = contextCarrier.items();
-        while (next.hasNext()) {
-            next = next.next();
-            clientHttpRequest.getHeaders().set(next.getHeadKey(), next.getHeadValue());
+        if (contextCarrier != null) {
+            inject(clientHttpRequest, contextCarrier);
         }
     }
 
     @Override
     public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
                               Object ret) throws Throwable {
-        return ret;
+        // Connectors like JdkClientHttpConnector invoke writeTo eagerly at assembly time,
+        // before the exchange interceptor sets the carrier at subscription time. Retry the
+        // injection when the returned Mono is subscribed, before the request is committed.
+        if (objInst.getSkyWalkingDynamicField() != null || !(ret instanceof Mono)) {
+            return ret;
+        }
+        final ClientHttpRequest clientHttpRequest = (ClientHttpRequest) allArguments[0];
+        return Mono.defer(() -> {
+            ContextCarrier contextCarrier = (ContextCarrier) objInst.getSkyWalkingDynamicField();
+            if (contextCarrier != null) {
+                try {
+                    inject(clientHttpRequest, contextCarrier);
+                } catch (Throwable t) {
+                    // headers are read-only once the request is committed (e.g. re-subscribed by a retry)
+                }
+            }
+            return (Mono<?>) ret;
+        });
+    }
+
+    private void inject(ClientHttpRequest clientHttpRequest, ContextCarrier contextCarrier) {
+        CarrierItem next = contextCarrier.items();
+        while (next.hasNext()) {
+            next = next.next();
+            clientHttpRequest.getHeaders().set(next.getHeadKey(), next.getHeadValue());
+        }
     }
 
     @Override