From 0c4707128250f5cb0276ae55b4665a230ef22668 Mon Sep 17 00:00:00 2001 From: eye-gu Date: Tue, 22 Sep 2026 15:05:41 +0800 Subject: [PATCH 1/2] Fix the WebClient plugins' NPE when writeTo runs before the exit span is created --- CHANGES.md | 6 ++++ .../BodyInserterRequestInterceptor.java | 30 +++++++++++++++---- .../BodyInserterRequestInterceptor.java | 30 +++++++++++++++---- 3 files changed, 56 insertions(+), 10 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index a175738df6..aba888c782 100644 --- a/CHANGES.md +++ b/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). diff --git a/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 b/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 index f65d7c06a1..030532d01f 100644 --- a/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 +++ b/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,36 @@ 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) { + inject(clientHttpRequest, contextCarrier); + } + 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 diff --git a/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 b/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 index 47bb3f610c..db68516fe1 100644 --- a/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 +++ b/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,36 @@ 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) { + inject(clientHttpRequest, contextCarrier); + } + 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 From 7e13989af5284185ba75a80c97763d5ad958cdc6 Mon Sep 17 00:00:00 2001 From: eye-gu Date: Tue, 22 Sep 2026 17:04:37 +0800 Subject: [PATCH 2/2] Swallow failures of the deferred carrier injection on re-subscription When the exchange Mono is re-subscribed, e.g. by a retry in an ExchangeFilterFunction, the deferred injection re-runs against the same already committed ClientHttpRequest whose headers are read-only, and the UnsupportedOperationException propagates into the application's reactive chain and fails the request. Catch it in both the 5.x and 6.x interceptors, as suggested in the review of #834: the sw8 header injected on the first subscription is still sent on the retried request. --- .../v5/webclient/BodyInserterRequestInterceptor.java | 6 +++++- .../v6/webclient/BodyInserterRequestInterceptor.java | 6 +++++- 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/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 b/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 index 030532d01f..d3772afb5d 100644 --- a/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 +++ b/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 @@ -53,7 +53,11 @@ public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allA return Mono.defer(() -> { ContextCarrier contextCarrier = (ContextCarrier) objInst.getSkyWalkingDynamicField(); if (contextCarrier != null) { - inject(clientHttpRequest, contextCarrier); + 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; }); diff --git a/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 b/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 index db68516fe1..9765e4ef77 100644 --- a/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 +++ b/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 @@ -53,7 +53,11 @@ public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allA return Mono.defer(() -> { ContextCarrier contextCarrier = (ContextCarrier) objInst.getSkyWalkingDynamicField(); if (contextCarrier != null) { - inject(clientHttpRequest, contextCarrier); + 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; });