diff --git a/CHANGES.md b/CHANGES.md
index 79a40e725d..3f30b4f687 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -4,6 +4,7 @@ Release Notes.
9.8.0
------------------
+* Fix `httpclient-5.x-plugin` closing the caller thread's active span when `FutureCallback` executes on the caller thread (apache/skywalking#14097).
* Fix the `NullPointerException` thrown by the `spring-webflux-5.x-webclient` and
`spring-webflux-6.x-webclient` plugins when `DefaultClientRequestBuilder$BodyInserterRequest#writeTo` runs
diff --git a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/AsyncRequestSpans.java b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/AsyncRequestSpans.java
new file mode 100644
index 0000000000..dfb24da289
--- /dev/null
+++ b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/AsyncRequestSpans.java
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpclient.v5;
+
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.hc.core5.http.HttpHost;
+import org.apache.skywalking.apm.agent.core.context.tag.Tags;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
+
+/**
+ * Per-request async exit span, owned by the request itself rather than by whatever thread happens to be running
+ * when a callback fires.
+ *
+ *
The span is created once, on the caller thread inside {@code doExecute}, while the caller's tracing context is
+ * still active. It is then immediately detached via {@link AbstractSpan#prepareForAsync()} +
+ * {@code ContextManager.stopSpan(span)} so it never sits on any thread's active-span stack while the request is in
+ * flight. From that point on it is finished exactly once, by reference, from whichever lifecycle callback gets
+ * there first (I/O thread response consumer, or the future callback on the caller/business thread) — never by a
+ * parameterless {@code ContextManager.stopSpan()} that would blindly pop whatever span is currently active on that
+ * thread.
+ *
+ *
All mutating operations are synchronized: {@link #onResponse(int)} (tagging, typically the I/O thread) can
+ * otherwise race with {@link #finish()} / {@link #fail(Throwable)} (typically the response-consumer or callback
+ * thread) finishing and clearing the span in the same window. An {@link AtomicReference} alone would prevent a
+ * double-finish but not a tag-write racing a finish.
+ */
+public class AsyncRequestSpans {
+
+ private final HttpHost target;
+
+ /**
+ * Only true, and only once, on the thread that is still inside {@code doExecute} when the request producer
+ * hands the concrete request to the channel. Any other thread (a custom {@code AsyncRequestProducer} that
+ * defers sending) has no relationship to the caller's context, so it must not create a span.
+ */
+ private final AtomicReference creator = new AtomicReference<>(Thread.currentThread());
+
+ private AbstractSpan span;
+ private boolean finished;
+
+ public AsyncRequestSpans(HttpHost target) {
+ this.target = target;
+ }
+
+ public HttpHost getTarget() {
+ return target;
+ }
+
+ /**
+ * Claims the right to create the span. Returns {@code true} at most once, and only for the thread that
+ * constructed this holder (the {@code doExecute} caller thread).
+ */
+ public boolean claimCreation() {
+ Thread current = Thread.currentThread();
+ return creator.compareAndSet(current, null);
+ }
+
+ /**
+ * Called at the end of {@code doExecute} (success or failure) so a late/duplicate send from the same thread
+ * cannot still claim creation after the caller has moved on.
+ */
+ public void callerReturned() {
+ creator.set(null);
+ }
+
+ /**
+ * Stores the span. Must be called only after the span has already been detached with
+ * {@code prepareForAsync()} + {@code ContextManager.stopSpan(span)} — this class never touches the active-span
+ * stack itself.
+ */
+ public synchronized void start(AbstractSpan span) {
+ this.span = span;
+ }
+
+ public synchronized void onResponse(int statusCode) {
+ if (span == null || finished) {
+ return;
+ }
+
+ Tags.HTTP_RESPONSE_STATUS_CODE.set(span, statusCode);
+
+ if (statusCode >= 400) {
+ span.errorOccurred();
+ }
+ }
+
+ /** The whole response completed successfully. */
+ public synchronized void finish() {
+ end(false, null);
+ }
+
+ /** The exchange failed with an exception. */
+ public synchronized void fail(Throwable cause) {
+ end(true, cause);
+ }
+
+ /**
+ * Cancelled, or resources released before the response ever completed (e.g. a redirect exec that declines to
+ * resend a non-repeatable entity and never invokes {@code completed()}). Only takes effect if the span is
+ * still open — the normal-completion paths already finished it earlier, so this is then a no-op.
+ */
+ public synchronized void abort() {
+ end(true, null);
+ }
+
+ private void end(boolean error, Throwable cause) {
+ if (span == null || finished) {
+ return;
+ }
+
+ finished = true;
+
+ if (error) {
+ span.errorOccurred();
+ }
+
+ if (cause != null) {
+ span.log(cause);
+ }
+
+ span.asyncFinish();
+ span = null;
+ }
+}
\ No newline at end of file
diff --git a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/HttpAsyncClientDoExecuteInterceptor.java b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/HttpAsyncClientDoExecuteInterceptor.java
index 68267fcc6f..8e8fb92fc8 100644
--- a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/HttpAsyncClientDoExecuteInterceptor.java
+++ b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/HttpAsyncClientDoExecuteInterceptor.java
@@ -18,42 +18,93 @@
package org.apache.skywalking.apm.plugin.httpclient.v5;
+import java.lang.reflect.Method;
import org.apache.hc.core5.concurrent.FutureCallback;
+import org.apache.hc.core5.http.HttpHost;
+import org.apache.hc.core5.http.nio.AsyncRequestProducer;
import org.apache.hc.core5.http.nio.AsyncResponseConsumer;
-import org.apache.hc.core5.http.protocol.HttpContext;
import org.apache.skywalking.apm.agent.core.context.ContextManager;
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.plugin.httpclient.v5.wrapper.AsyncRequestProducerWrapper;
import org.apache.skywalking.apm.plugin.httpclient.v5.wrapper.AsyncResponseConsumerWrapper;
import org.apache.skywalking.apm.plugin.httpclient.v5.wrapper.FutureCallbackWrapper;
-import java.lang.reflect.Method;
-
+/**
+ * Intercepts the internal {@code doExecute(HttpHost, AsyncRequestProducer, AsyncResponseConsumer, ..., FutureCallback)}
+ * overload shared by every async client implementation (Internal*AsyncClient, Minimal*AsyncClient, and the
+ * classic-facade adapter), whose argument order/types are identical across HttpClient 5.0 through 5.6.
+ *
+ *
Unlike the previous implementation, this interceptor never stores anything in the {@code HttpContext} and
+ * never wraps a callback purely to call a parameterless {@code ContextManager.stopSpan()}. It only:
+ *
+ *
creates a per-request {@link AsyncRequestSpans} holder, while the caller's context is still active;
+ *
wraps the request producer so the exit span is created on the caller thread, synchronously, the moment the
+ * concrete {@code HttpRequest} becomes available;
+ *
wraps the response consumer and future callback so the retained span is finished by reference.
+ *
+ * Because span creation no longer depends on the {@code HttpContext}, this also fixes HttpClient 5.4+, where the
+ * context argument passed by the classic facade and by {@code execute(SimpleHttpRequest, FutureCallback)} is
+ * {@code null}.
+ */
public class HttpAsyncClientDoExecuteInterceptor implements InstanceMethodsAroundInterceptor {
+ private static final int TARGET_INDEX = 0;
+ private static final int REQUEST_PRODUCER_INDEX = 1;
+ private static final int RESPONSE_CONSUMER_INDEX = 2;
+ private static final int CALLBACK_INDEX = 5;
+
@Override
- public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes,
- MethodInterceptResult result) throws Throwable {
- AsyncResponseConsumer consumer = (AsyncResponseConsumer) allArguments[2];
- HttpContext context = (HttpContext) allArguments[4];
- FutureCallback callback = (FutureCallback) allArguments[5];
- allArguments[2] = new AsyncResponseConsumerWrapper(consumer);
- allArguments[5] = new FutureCallbackWrapper(callback);
- if (ContextManager.isActive()) {
- context.setAttribute(Constants.SKYWALKING_CONTEXT_SNAPSHOT, ContextManager.capture());
+ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+ Class[] argumentsTypes, MethodInterceptResult result) throws Throwable {
+ if (!ContextManager.isActive()) {
+ return;
}
+ if (!(allArguments[REQUEST_PRODUCER_INDEX] instanceof AsyncRequestProducer)
+ || !(allArguments[RESPONSE_CONSUMER_INDEX] instanceof AsyncResponseConsumer)) {
+ return;
+ }
+
+ final HttpHost target = allArguments[TARGET_INDEX] instanceof HttpHost
+ ? (HttpHost) allArguments[TARGET_INDEX] : null;
+ final AsyncRequestSpans spans = new AsyncRequestSpans(target);
+
+ allArguments[REQUEST_PRODUCER_INDEX] = new AsyncRequestProducerWrapper(
+ (AsyncRequestProducer) allArguments[REQUEST_PRODUCER_INDEX], spans);
+ allArguments[RESPONSE_CONSUMER_INDEX] = new AsyncResponseConsumerWrapper<>(
+ (AsyncResponseConsumer) allArguments[RESPONSE_CONSUMER_INDEX], spans);
+ // Wrap even when the caller passed null: it's the only lifecycle hook that sees cancellation and the
+ // synchronous-failure-before-consumer-runs path for callers who supplied no callback of their own.
+ allArguments[CALLBACK_INDEX] = new FutureCallbackWrapper<>(
+ (FutureCallback) allArguments[CALLBACK_INDEX], spans);
}
@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) throws Throwable {
+ releaseCreationClaim(allArguments);
return ret;
}
@Override
public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
- Class[] argumentsTypes, Throwable t) {
+ Class[] argumentsTypes, Throwable t) {
+ if (allArguments[REQUEST_PRODUCER_INDEX] instanceof AsyncRequestProducerWrapper) {
+ AsyncRequestProducerWrapper wrapper = (AsyncRequestProducerWrapper) allArguments[REQUEST_PRODUCER_INDEX];
+ wrapper.getSpans().fail(t);
+ }
+ releaseCreationClaim(allArguments);
+ }
+ /**
+ * Once {@code doExecute} has returned (or thrown), no thread other than a genuinely deferred custom producer
+ * has any business claiming span creation — clearing this here keeps {@link AsyncRequestSpans#claimCreation()}
+ * honest even if the same thread somehow re-enters.
+ */
+ private void releaseCreationClaim(Object[] allArguments) {
+ if (allArguments[REQUEST_PRODUCER_INDEX] instanceof AsyncRequestProducerWrapper) {
+ ((AsyncRequestProducerWrapper) allArguments[REQUEST_PRODUCER_INDEX]).getSpans().callerReturned();
+ }
}
}
diff --git a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/IOSessionImplPollInterceptor.java b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/IOSessionImplPollInterceptor.java
deleted file mode 100644
index fc8ef190d4..0000000000
--- a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/IOSessionImplPollInterceptor.java
+++ /dev/null
@@ -1,92 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package org.apache.skywalking.apm.plugin.httpclient.v5;
-
-import org.apache.hc.client5.http.protocol.HttpClientContext;
-import org.apache.hc.core5.http.message.BasicHttpRequest;
-import org.apache.hc.core5.http.nio.command.RequestExecutionCommand;
-import org.apache.hc.core5.http.protocol.HttpContext;
-import org.apache.hc.core5.reactor.Command;
-import org.apache.skywalking.apm.agent.core.context.CarrierItem;
-import org.apache.skywalking.apm.agent.core.context.ContextCarrier;
-import org.apache.skywalking.apm.agent.core.context.ContextManager;
-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.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
-import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
-import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
-
-import java.lang.reflect.Method;
-import java.net.URI;
-
-public class IOSessionImplPollInterceptor implements InstanceMethodsAroundInterceptor {
-
- @Override
- public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes,
- MethodInterceptResult result) throws Throwable {
-
- }
-
- @Override
- public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes,
- Object ret) throws Throwable {
- Command command = (Command) ret;
- if (!(command instanceof RequestExecutionCommand)) {
- return ret;
- }
- HttpContext httpContext = ((RequestExecutionCommand) command).getContext();
- ContextSnapshot snapshot = (ContextSnapshot) httpContext.getAttribute(Constants.SKYWALKING_CONTEXT_SNAPSHOT);
- if (snapshot == null) {
- return ret;
- }
- httpContext.removeAttribute(Constants.SKYWALKING_CONTEXT_SNAPSHOT);
- AbstractSpan localSpan = ContextManager.createLocalSpan("httpasyncclient/local");
- localSpan.setComponent(ComponentsDefine.HTTP_ASYNC_CLIENT);
- localSpan.setLayer(SpanLayer.HTTP);
- ContextManager.continued(snapshot);
-
- final ContextCarrier contextCarrier = new ContextCarrier();
- BasicHttpRequest request = (BasicHttpRequest) httpContext.getAttribute(HttpClientContext.HTTP_REQUEST);
- URI uri = request.getUri();
-
- String operationName = uri.getPath();
- int port = uri.getPort();
- AbstractSpan span = ContextManager
- .createExitSpan(operationName, contextCarrier, uri.getHost() + ":" + (port == -1 ? 80 : port));
- span.setComponent(ComponentsDefine.HTTP_ASYNC_CLIENT);
- Tags.URL.set(span, uri.toURL().toString());
- Tags.HTTP.METHOD.set(span, request.getMethod());
- SpanLayer.asHttp(span);
- CarrierItem next = contextCarrier.items();
- while (next.hasNext()) {
- next = next.next();
- request.setHeader(next.getHeadKey(), next.getHeadValue());
- }
- return ret;
- }
-
- @Override
- public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
- Class[] argumentsTypes, Throwable t) {
-
- }
-}
diff --git a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/define/IOSessionImplInstrumentation.java b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/define/IOSessionImplInstrumentation.java
deleted file mode 100644
index 5841089101..0000000000
--- a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/define/IOSessionImplInstrumentation.java
+++ /dev/null
@@ -1,68 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package org.apache.skywalking.apm.plugin.httpclient.v5.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.enhance.ClassInstanceMethodsEnhancePluginDefine;
-import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
-
-import static net.bytebuddy.matcher.ElementMatchers.named;
-import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName;
-
-public class IOSessionImplInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {
-
- private static final String ENHANCE_CLASS = "org.apache.hc.core5.reactor.IOSessionImpl";
- private static final String METHOD_NAME = "poll";
- private static final String INTERCEPT_CLASS = "org.apache.skywalking.apm.plugin.httpclient.v5.IOSessionImplPollInterceptor";
-
- @Override
- protected ClassMatch enhanceClass() {
- return byName(ENHANCE_CLASS);
- }
-
- @Override
- public ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
- return null;
- }
-
- @Override
- public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
- return new InstanceMethodsInterceptPoint[]{
- new InstanceMethodsInterceptPoint() {
- @Override
- public ElementMatcher getMethodsMatcher() {
- return named(METHOD_NAME);
- }
-
- @Override
- public String getMethodsInterceptor() {
- return INTERCEPT_CLASS;
- }
-
- @Override
- public boolean isOverrideArgs() {
- return false;
- }
- }
- };
- }
-}
diff --git a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/AsyncRequestProducerWrapper.java b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/AsyncRequestProducerWrapper.java
new file mode 100644
index 0000000000..f355180657
--- /dev/null
+++ b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/AsyncRequestProducerWrapper.java
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpclient.v5.wrapper;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import org.apache.hc.core5.http.HttpException;
+import org.apache.hc.core5.http.HttpHost;
+import org.apache.hc.core5.http.HttpRequest;
+import org.apache.hc.core5.http.nio.AsyncRequestProducer;
+import org.apache.hc.core5.http.nio.DataStreamChannel;
+import org.apache.hc.core5.http.nio.RequestChannel;
+import org.apache.hc.core5.http.protocol.HttpContext;
+import org.apache.skywalking.apm.agent.core.context.CarrierItem;
+import org.apache.skywalking.apm.agent.core.context.ContextCarrier;
+import org.apache.skywalking.apm.agent.core.context.ContextManager;
+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.logging.api.ILog;
+import org.apache.skywalking.apm.agent.core.logging.api.LogManager;
+import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
+import org.apache.skywalking.apm.plugin.httpclient.v5.AsyncRequestSpans;
+
+/**
+ * Delegates every {@link AsyncRequestProducer} method unchanged, except {@link #sendRequest}, where it wraps the
+ * {@link RequestChannel} the underlying producer is handed. All standard producers (Internal/Minimal async
+ * clients, and the classic-facade adapter) call {@code channel.sendRequest(...)} synchronously, on the calling
+ * thread, from inside {@code doExecute} — so this is where the concrete {@link HttpRequest} first becomes
+ * available, while the caller's tracing context is still active.
+ */
+public class AsyncRequestProducerWrapper implements AsyncRequestProducer {
+
+ private static final ILog LOGGER = LogManager.getLogger(AsyncRequestProducerWrapper.class);
+
+ private final AsyncRequestProducer producer;
+ private final AsyncRequestSpans spans;
+
+ public AsyncRequestProducerWrapper(AsyncRequestProducer producer, AsyncRequestSpans spans) {
+ this.producer = producer;
+ this.spans = spans;
+ }
+
+ public AsyncRequestSpans getSpans() {
+ return spans;
+ }
+
+ @Override
+ public void sendRequest(RequestChannel channel, HttpContext context) throws HttpException, IOException {
+ producer.sendRequest((request, entityDetails, ctx) -> {
+ if (spans.claimCreation()) {
+ try {
+ startExitSpan(request);
+ } catch (Throwable t) {
+ // Tracing must never break the user's actual HTTP request.
+ LOGGER.error(t, "Failed to trace the async HttpClient request.");
+ }
+ }
+ channel.sendRequest(request, entityDetails, ctx);
+ }, context);
+ }
+
+ private void startExitSpan(HttpRequest request) throws URISyntaxException {
+ URI uri = request.getUri();
+ HttpHost target = spans.getTarget();
+ // Same precedence InternalAbstractHttpAsyncClient itself uses: an explicit target host wins over
+ // whatever authority happens to be on the request URI.
+ String scheme = target != null ? target.getSchemeName() : uri.getScheme();
+ String host = target != null ? target.getHostName() : uri.getHost();
+ int port = target != null ? target.getPort() : uri.getPort();
+ if (host == null) {
+ return;
+ }
+ if (scheme == null) {
+ scheme = "http";
+ }
+ if (port < 0) {
+ port = "https".equalsIgnoreCase(scheme) ? 443 : 80;
+ }
+ String peer = host + ":" + port;
+ String path = uri.getPath() == null || uri.getPath().isEmpty() ? "/" : uri.getPath();
+ String url = scheme + "://" + peer + path + (uri.getRawQuery() == null ? "" : "?" + uri.getRawQuery());
+
+ // If we're already inside another plugin's exit span, createExitSpan reuses that span (nested depth + 1)
+ // instead of creating a new one. We must not treat a reused outer span as ours to detach/finish
+ // asynchronously — that lifecycle belongs to whichever plugin created it.
+ boolean nested = ContextManager.activeSpan() != null && ContextManager.activeSpan().isExit();
+
+ ContextCarrier carrier = new ContextCarrier();
+ AbstractSpan span = ContextManager.createExitSpan(path, carrier, peer);
+ try {
+ if (!nested) {
+ span.setComponent(ComponentsDefine.HTTP_ASYNC_CLIENT);
+ Tags.URL.set(span, url);
+ Tags.HTTP.METHOD.set(span, request.getMethod());
+ SpanLayer.asHttp(span);
+ }
+ CarrierItem next = carrier.items();
+ while (next.hasNext()) {
+ next = next.next();
+ request.setHeader(next.getHeadKey(), next.getHeadValue());
+ }
+ } finally {
+ if (!nested) {
+ // Detach BEFORE returning control to the channel: the client can report a synchronous failure
+ // back to doExecute's own catch block on this very thread before sendRequest() returns.
+ span.prepareForAsync();
+ ContextManager.stopSpan(span);
+ spans.start(span);
+ } else {
+ ContextManager.stopSpan(span);
+ }
+ }
+ }
+
+ @Override
+ public void failed(Exception cause) {
+ producer.failed(cause);
+ }
+
+ @Override
+ public boolean isRepeatable() {
+ return producer.isRepeatable();
+ }
+
+ @Override
+ public void produce(DataStreamChannel channel) throws IOException {
+ producer.produce(channel);
+ }
+
+ @Override
+ public int available() {
+ return producer.available();
+ }
+
+ @Override
+ public void releaseResources() {
+ producer.releaseResources();
+ }
+
+ // NOTE FOR AYUSH: AsyncRequestProducer's exact method set has drifted slightly across httpcore5 minor
+ // versions (5.0 vs 5.3+). Let your IDE's "implement remaining interface methods" fill in anything missing
+ // here (there should be none beyond the above in 5.0-5.6, but verify against the version this module
+ // actually compiles against) — every one of them should be a plain one-line delegate to `producer`, same
+ // as above. The only method with real logic is sendRequest().
+}
diff --git a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/AsyncResponseConsumerWrapper.java b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/AsyncResponseConsumerWrapper.java
index 9dec7d109a..b8218f3ead 100644
--- a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/AsyncResponseConsumerWrapper.java
+++ b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/AsyncResponseConsumerWrapper.java
@@ -13,69 +13,62 @@
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
+ *
*/
package org.apache.skywalking.apm.plugin.httpclient.v5.wrapper;
-import org.apache.hc.core5.concurrent.FutureCallback;
+import java.io.IOException;
import org.apache.hc.core5.http.EntityDetails;
-import org.apache.hc.core5.http.Header;
import org.apache.hc.core5.http.HttpException;
import org.apache.hc.core5.http.HttpResponse;
import org.apache.hc.core5.http.nio.AsyncResponseConsumer;
import org.apache.hc.core5.http.nio.CapacityChannel;
import org.apache.hc.core5.http.protocol.HttpContext;
-import org.apache.skywalking.apm.agent.core.context.ContextManager;
-import org.apache.skywalking.apm.agent.core.context.tag.Tags;
-import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
-
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.List;
+import org.apache.skywalking.apm.plugin.httpclient.v5.AsyncRequestSpans;
+/**
+ * Runs entirely on the I/O thread (with the sole exception that {@code releaseResources()} can also be invoked
+ * from elsewhere during cleanup). Never touches {@code ContextManager}'s active-span stack — only ever tags or
+ * finishes {@link #spans} by reference, which is safe to do from any thread.
+ */
public class AsyncResponseConsumerWrapper implements AsyncResponseConsumer {
- private AsyncResponseConsumer consumer;
+ private final AsyncResponseConsumer consumer;
+ private final AsyncRequestSpans spans;
- public AsyncResponseConsumerWrapper(AsyncResponseConsumer consumer) {
+ public AsyncResponseConsumerWrapper(AsyncResponseConsumer consumer, AsyncRequestSpans spans) {
this.consumer = consumer;
+ this.spans = spans;
}
@Override
public void consumeResponse(HttpResponse response, EntityDetails entityDetails, HttpContext context,
- FutureCallback resultCallback) throws HttpException, IOException {
- if (ContextManager.isActive()) {
- int statusCode = response.getCode();
- AbstractSpan span = ContextManager.activeSpan();
- Tags.HTTP_RESPONSE_STATUS_CODE.set(span, statusCode);
- if (statusCode >= 400) {
- span.errorOccurred();
- }
- ContextManager.stopSpan();
+ org.apache.hc.core5.concurrent.FutureCallback resultCallback) throws HttpException, IOException {
+ spans.onResponse(response.getCode());
+ if (entityDetails == null) {
+ // No body means streamEnd() will never be called for this exchange.
+ spans.finish();
}
consumer.consumeResponse(response, entityDetails, context, resultCallback);
}
@Override
public void informationResponse(HttpResponse response, HttpContext context) throws HttpException, IOException {
- if (ContextManager.isActive()) {
- int statusCode = response.getCode();
- AbstractSpan span = ContextManager.activeSpan();
- Tags.HTTP_RESPONSE_STATUS_CODE.set(span, statusCode);
- if (statusCode >= 400) {
- span.errorOccurred();
- }
- ContextManager.stopSpan();
- }
+ // 1xx is not the final response; the exit span's status must come from the final consumeResponse() call.
consumer.informationResponse(response, context);
}
+ @Override
+ public void streamEnd(java.util.List trailers)
+ throws HttpException, IOException {
+ spans.finish();
+ consumer.streamEnd(trailers);
+ }
+
@Override
public void failed(Exception cause) {
- if (ContextManager.isActive()) {
- ContextManager.activeSpan().errorOccurred().log(cause);
- ContextManager.stopSpan();
- }
+ spans.fail(cause);
consumer.failed(cause);
}
@@ -85,17 +78,25 @@ public void updateCapacity(CapacityChannel capacityChannel) throws IOException {
}
@Override
- public void consume(ByteBuffer src) throws IOException {
+ public void consume(java.nio.ByteBuffer src) throws IOException {
consumer.consume(src);
}
- @Override
- public void streamEnd(List trailers) throws HttpException, IOException {
- consumer.streamEnd(trailers);
- }
-
@Override
public void releaseResources() {
+ // Fallback finisher, not a success signal: HttpAsyncMainClientExec#failed calls releaseResources()
+ // *before* reporting the failure, and a suppressed-redirect-with-non-repeatable-entity exchange only
+ // ever calls completed() without a real response. abort() only takes effect if the span is still open —
+ // every normal-completion path above has already finished it by the time release runs, so this is then
+ // a no-op. If the span IS still open here, the exchange ended without a complete response, so it's
+ // correctly marked as an error rather than silently dropped.
+ spans.abort();
consumer.releaseResources();
}
+
+ // NOTE FOR AYUSH: same caveat as AsyncRequestProducerWrapper — let the IDE fill in any interface method not
+ // listed above (e.g. some httpcore5 versions' AsyncResponseConsumer exposes it slightly differently); every
+ // one you add should be a plain delegate to `consumer` with zero span logic. The five methods above
+ // (consumeResponse, informationResponse, streamEnd, failed, releaseResources) are the only ones that matter
+ // for span lifecycle.
}
diff --git a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/FutureCallbackWrapper.java b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/FutureCallbackWrapper.java
index f606856edf..26ed80ac9e 100644
--- a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/FutureCallbackWrapper.java
+++ b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/FutureCallbackWrapper.java
@@ -13,48 +13,58 @@
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
+ *
*/
package org.apache.skywalking.apm.plugin.httpclient.v5.wrapper;
import org.apache.hc.core5.concurrent.FutureCallback;
-import org.apache.skywalking.apm.agent.core.context.ContextManager;
+import org.apache.skywalking.apm.plugin.httpclient.v5.AsyncRequestSpans;
+/**
+ * This is the class the original bug (#14097) lived in: the old implementation called the parameterless
+ * {@code ContextManager.stopSpan()} here, which pops whatever span is active on whatever thread happens to
+ * invoke this callback — and with {@code HttpAsyncClients.classic(...)}, that can be the caller/business
+ * thread, once it reads the response body to EOF. That thread's active span is the caller's own business span,
+ * not this HTTP request's span.
+ *
+ *
This version never touches the active-span stack. It only finishes {@link #spans} by reference, which is
+ * safe from any thread — the caller's own span is never at risk.
+ *
+ *
{@code completed}/{@code failed} are largely redundant with {@link AsyncResponseConsumerWrapper}'s own
+ * finish paths ({@link AsyncRequestSpans#finish()}/{@link AsyncRequestSpans#fail(Throwable)} are idempotent), but
+ * this callback still matters for {@link #cancelled()} — which the consumer never sees — and as a safety net for
+ * any exchange that completes without ever driving the consumer's normal lifecycle.
+ */
public class FutureCallbackWrapper implements FutureCallback {
- private FutureCallback callback;
+ private final FutureCallback callback;
+ private final AsyncRequestSpans spans;
- public FutureCallbackWrapper(FutureCallback callback) {
+ public FutureCallbackWrapper(FutureCallback callback, AsyncRequestSpans spans) {
this.callback = callback;
+ this.spans = spans;
}
@Override
- public void completed(T o) {
- if (ContextManager.isActive()) {
- ContextManager.stopSpan();
- }
+ public void completed(T result) {
+ spans.finish();
if (callback != null) {
- callback.completed(o);
+ callback.completed(result);
}
}
@Override
- public void failed(Exception e) {
- if (ContextManager.isActive()) {
- ContextManager.activeSpan().errorOccurred().log(e);
- ContextManager.stopSpan();
- }
+ public void failed(Exception ex) {
+ spans.fail(ex);
if (callback != null) {
- callback.failed(e);
+ callback.failed(ex);
}
}
@Override
public void cancelled() {
- if (ContextManager.isActive()) {
- ContextManager.activeSpan().errorOccurred();
- ContextManager.stopSpan();
- }
+ spans.abort();
if (callback != null) {
callback.cancelled();
}
diff --git a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/resources/skywalking-plugin.def b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/resources/skywalking-plugin.def
index dc6622a88f..63c6348953 100644
--- a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/resources/skywalking-plugin.def
+++ b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/resources/skywalking-plugin.def
@@ -17,4 +17,3 @@
httpclient-5.x=org.apache.skywalking.apm.plugin.httpclient.v5.define.MinimalHttpClientInstrumentation
httpclient-5.x=org.apache.skywalking.apm.plugin.httpclient.v5.define.InternalHttpClientInstrumentation
httpclient-5.x=org.apache.skywalking.apm.plugin.httpclient.v5.define.HttpAsyncClientInstrumentation
-httpclient-5.x=org.apache.skywalking.apm.plugin.httpclient.v5.define.IOSessionImplInstrumentation
diff --git a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/httpclient/v5/AsyncRequestSpansTest.java b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/httpclient/v5/AsyncRequestSpansTest.java
new file mode 100644
index 0000000000..f1d6ba8aeb
--- /dev/null
+++ b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/httpclient/v5/AsyncRequestSpansTest.java
@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpclient.v5;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnit;
+import org.mockito.junit.MockitoRule;
+import org.junit.Rule;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+/**
+ * These tests intentionally don't touch ContextManager/ByteBuddy at all — {@link AsyncRequestSpans} owns no
+ * thread-stack state, so its lifecycle guarantees (exactly-once finish, correct error propagation, claim
+ * exclusivity) can and should be verified directly, without a TracingSegmentRunner. Thread-stack correctness
+ * (nothing leaks onto the reactor thread, the caller's own span survives) belongs in the plugin scenario, not
+ * here — see test/plugin/scenarios/httpclient-5.x-scenario.
+ */
+public class AsyncRequestSpansTest {
+
+ @Rule
+ public MockitoRule mockitoRule = MockitoJUnit.rule();
+
+ @Mock
+ private AbstractSpan span;
+
+ private AsyncRequestSpans spans;
+
+ @Before
+ public void setUp() {
+ spans = new AsyncRequestSpans(null);
+ spans.start(span);
+ }
+
+ @Test
+ public void finishIsAppliedExactlyOnce() {
+ spans.finish();
+ spans.finish();
+ spans.fail(new RuntimeException("late failure after already finished"));
+
+ verify(span, times(1)).asyncFinish();
+ }
+
+ @Test
+ public void streamEndThenReleaseResourcesDoesNotDoubleFinishOrMarkError() {
+ // consumeResponse (no error status) -> streamEnd -> releaseResources, the normal successful path.
+ spans.onResponse(200);
+ spans.finish();
+ spans.abort(); // what releaseResources() calls; must be a no-op once already finished
+
+ verify(span, times(1)).asyncFinish();
+ verify(span, never()).errorOccurred();
+ }
+
+ @Test
+ public void releaseResourcesBeforeFailedStillEndsAsError() {
+ // HttpAsyncMainClientExec#failed calls releaseResources() BEFORE reporting the failure. If the span is
+ // still open when releaseResources() runs, the exchange never completed successfully, so it must be
+ // marked an error even though `fail()` with the real cause hasn't been called yet.
+ spans.onResponse(200);
+ spans.abort(); // releaseResources() fires first, response never fully arrived
+
+ verify(span, times(1)).errorOccurred();
+ verify(span, times(1)).asyncFinish();
+ }
+
+ @Test
+ public void noBodyFinishesAtConsumeResponse() {
+ spans.onResponse(204); // no entity -> caller calls finish() directly, streamEnd() never comes
+ spans.finish();
+
+ verify(span, times(1)).asyncFinish();
+ }
+
+ @Test
+ public void bodyFailureAfterSuccessfulHeadersIsAnError() {
+ spans.onResponse(200);
+ spans.fail(new RuntimeException("body read failed"));
+
+ verify(span, times(1)).errorOccurred();
+ verify(span, times(1)).log(org.mockito.ArgumentMatchers.any(Throwable.class));
+ verify(span, times(1)).asyncFinish();
+ }
+
+ @Test
+ public void cancellationEndsAsError() {
+ spans.abort();
+
+ verify(span, times(1)).errorOccurred();
+ verify(span, times(1)).asyncFinish();
+ }
+
+ @Test
+ public void errorStatusCodeMarksErrorWithoutFinishing() {
+ spans.onResponse(500);
+
+ verify(span, times(1)).errorOccurred();
+ verify(span, never()).asyncFinish();
+ }
+
+ @Test
+ public void informationResponseDoesNotFinishOrTagStatus() {
+ // 1xx must be a pure passthrough at the wrapper level; AsyncRequestSpans is simply never called for it.
+ // Nothing to assert here beyond "no interaction" — covered by not invoking onResponse/finish at all.
+ verify(span, never()).asyncFinish();
+ }
+
+ @Test
+ public void onlyTheCreatingThreadCanClaimCreation() throws InterruptedException {
+ AsyncRequestSpans fresh = new AsyncRequestSpans(null);
+ AtomicInteger claims = new AtomicInteger();
+ CountDownLatch done = new CountDownLatch(1);
+
+ // A different thread -- standing in for a custom AsyncRequestProducer that defers sending to another
+ // thread -- must NOT be able to claim creation. Only the constructing (doExecute) thread may.
+ new Thread(() -> {
+ if (fresh.claimCreation()) {
+ claims.incrementAndGet();
+ }
+ done.countDown();
+ }).start();
+ assertTrue(done.await(5, TimeUnit.SECONDS));
+ assertEquals(0, claims.get());
+
+ assertTrue(fresh.claimCreation());
+ assertEquals(false, fresh.claimCreation()); // exactly once, even for the right thread
+ }
+
+ @Test
+ public void callerReturnedRevokesClaimEvenIfUnused() {
+ AsyncRequestSpans fresh = new AsyncRequestSpans(null);
+ fresh.callerReturned();
+
+ assertEquals(false, fresh.claimCreation());
+ }
+}
diff --git a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/AsyncRequestProducerWrapperTest.java b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/AsyncRequestProducerWrapperTest.java
new file mode 100644
index 0000000000..f119df5193
--- /dev/null
+++ b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/AsyncRequestProducerWrapperTest.java
@@ -0,0 +1,218 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpclient.v5.wrapper;
+
+import java.net.URI;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.apache.hc.core5.http.HttpHost;
+import org.apache.hc.core5.http.HttpRequest;
+import org.apache.hc.core5.http.message.BasicHttpRequest;
+import org.apache.hc.core5.http.nio.AsyncRequestProducer;
+import org.apache.hc.core5.http.nio.RequestChannel;
+import org.apache.hc.core5.http.protocol.HttpContext;
+import org.apache.skywalking.apm.agent.core.context.ContextCarrier;
+import org.apache.skywalking.apm.agent.core.context.ContextManager;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
+import org.apache.skywalking.apm.agent.test.tools.AgentServiceRule;
+import org.apache.skywalking.apm.agent.test.tools.SegmentStorage;
+import org.apache.skywalking.apm.agent.test.tools.SegmentStoragePoint;
+import org.apache.skywalking.apm.agent.test.tools.TracingSegmentRunner;
+import org.apache.skywalking.apm.plugin.httpclient.v5.AsyncRequestSpans;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Exercises {@link AsyncRequestProducerWrapper} against the real {@link ContextManager}, via
+ * {@link TracingSegmentRunner}. {@code startExitSpan()} calls {@code ContextManager.createExitSpan},
+ * {@code AbstractSpan.prepareForAsync()} and {@code ContextManager.stopSpan()} directly — mocking those out
+ * would only prove a mock was invoked, not that the caller's own active-span stack is left correctly balanced,
+ * which is the entire point of this class (and of issue #14097).
+ *
+ *
{@code AsyncResponseConsumerWrapperTest} and {@code FutureCallbackWrapperTest} don't need this harness:
+ * neither ever touches {@code ContextManager} — only the {@link AsyncRequestSpans} reference they're handed.
+ *
+ *
Known gap, deliberate: there is no assertion here that the exit span's peer is built from the
+ * explicit target host rather than the request URI's authority. That would require reading a completed span
+ * back out of the archived {@code TraceSegment} (e.g. a peer accessor), and I don't have confirmed access to
+ * that accessor in this codebase — guessing it once already produced a compile failure, so I'm not guessing
+ * again. The target/URI precedence logic in {@code startExitSpan()} is a short, branch-free block that's easy
+ * to verify by reading it directly; if you tell me the actual read-side accessor (on whatever class
+ * {@code TraceSegment}/the span type actually exposes it), I'll add that assertion in a follow-up.
+ */
+@RunWith(TracingSegmentRunner.class)
+public class AsyncRequestProducerWrapperTest {
+
+ @SegmentStoragePoint
+ private SegmentStorage segmentStorage;
+
+ @Rule
+ public AgentServiceRule agentServiceRule = new AgentServiceRule();
+
+ private static final HttpHost TARGET = new HttpHost("http", "example.org", 8080);
+
+ /**
+ * Stands in for every real {@code AsyncRequestProducer} (Internal/Minimal async clients, classic-facade
+ * adapter): calls the {@link RequestChannel} it's handed synchronously, on the calling thread, with a
+ * concrete request — exactly what {@link AsyncRequestProducerWrapper#sendRequest} depends on.
+ */
+ private AsyncRequestProducer syncDelegate(HttpRequest request) throws Exception {
+ AsyncRequestProducer delegate = mock(AsyncRequestProducer.class);
+ doAnswer(invocation -> {
+ RequestChannel channel = invocation.getArgument(0);
+ HttpContext context = invocation.getArgument(1);
+ channel.sendRequest(request, null, context);
+ return null;
+ }).when(delegate).sendRequest(any(RequestChannel.class), any(HttpContext.class));
+ return delegate;
+ }
+
+ private HttpRequest requestTo(String uri) throws Exception {
+ return new BasicHttpRequest("GET", new URI(uri));
+ }
+
+ @Test
+ public void callerSpanRemainsActiveImmediatelyAfterHandoff() throws Exception {
+ AbstractSpan caller = ContextManager.createLocalSpan("caller");
+ AsyncRequestSpans spans = new AsyncRequestSpans(TARGET);
+ AsyncRequestProducerWrapper wrapper = new AsyncRequestProducerWrapper(
+ syncDelegate(requestTo("http://example.org/hello")), spans);
+
+ wrapper.sendRequest(mock(RequestChannel.class), mock(HttpContext.class));
+
+ assertSame(caller, ContextManager.activeSpan());
+
+ ContextManager.stopSpan(caller);
+ spans.finish();
+ }
+
+ @Test
+ public void exitSpanIsDetachedAndNotArchivedUntilAsyncFinish() throws Exception {
+ AbstractSpan outer = ContextManager.createLocalSpan("outer");
+ AsyncRequestSpans spans = new AsyncRequestSpans(TARGET);
+ AsyncRequestProducerWrapper wrapper = new AsyncRequestProducerWrapper(
+ syncDelegate(requestTo("http://example.org/hello")), spans);
+
+ wrapper.sendRequest(mock(RequestChannel.class), mock(HttpContext.class));
+ ContextManager.stopSpan(outer);
+
+ assertEquals(0, segmentStorage.getTraceSegments().size());
+
+ spans.finish();
+
+ assertEquals(1, segmentStorage.getTraceSegments().size());
+ }
+
+ @Test
+ public void headersAreInjectedIntoTheConcreteRequest() throws Exception {
+ AbstractSpan outer = ContextManager.createLocalSpan("outer");
+ HttpRequest request = requestTo("http://example.org/hello");
+ AsyncRequestSpans spans = new AsyncRequestSpans(TARGET);
+ AsyncRequestProducerWrapper wrapper = new AsyncRequestProducerWrapper(syncDelegate(request), spans);
+
+ wrapper.sendRequest(mock(RequestChannel.class), mock(HttpContext.class));
+
+ assertTrue("sw8 propagation header must be injected", request.containsHeader("sw8"));
+
+ ContextManager.stopSpan(outer);
+ spans.finish();
+ }
+
+ @Test
+ public void nestedInsideAnotherExitSpanDoesNotCreateASeparateAsyncSpan() throws Exception {
+ AbstractSpan outerExit = ContextManager.createExitSpan("outer-exit", new ContextCarrier(), "outer-peer:1");
+
+ AsyncRequestSpans spans = mock(AsyncRequestSpans.class);
+ when(spans.getTarget()).thenReturn(TARGET);
+ when(spans.claimCreation()).thenReturn(true);
+ AsyncRequestProducerWrapper wrapper = new AsyncRequestProducerWrapper(
+ syncDelegate(requestTo("http://example.org/hello")), spans);
+
+ wrapper.sendRequest(mock(RequestChannel.class), mock(HttpContext.class));
+
+ verify(spans, never()).start(any(AbstractSpan.class));
+ assertSame(outerExit, ContextManager.activeSpan());
+
+ ContextManager.stopSpan(outerExit);
+ }
+
+ @Test
+ public void sendRequestFromAnotherThreadNeverCreatesASpan() throws Exception {
+ AbstractSpan outer = ContextManager.createLocalSpan("outer");
+ AsyncRequestSpans spans = mock(AsyncRequestSpans.class);
+ when(spans.getTarget()).thenReturn(TARGET);
+ when(spans.claimCreation()).thenReturn(false);
+
+ HttpRequest request = requestTo("http://example.org/hello");
+ AsyncRequestProducer deferredDelegate = mock(AsyncRequestProducer.class);
+ CountDownLatch done = new CountDownLatch(1);
+ doAnswer(invocation -> {
+ RequestChannel channel = invocation.getArgument(0);
+ HttpContext context = invocation.getArgument(1);
+ Thread t = new Thread(() -> {
+ try {
+ channel.sendRequest(request, null, context);
+ } catch (Exception ignored) {
+ // test-only best effort
+ } finally {
+ done.countDown();
+ }
+ });
+ t.start();
+ return null;
+ }).when(deferredDelegate).sendRequest(any(RequestChannel.class), any(HttpContext.class));
+
+ AsyncRequestProducerWrapper wrapper = new AsyncRequestProducerWrapper(deferredDelegate, spans);
+ wrapper.sendRequest(mock(RequestChannel.class), mock(HttpContext.class));
+
+ assertTrue(done.await(5, TimeUnit.SECONDS));
+ verify(spans, never()).start(any(AbstractSpan.class));
+ assertSame(outer, ContextManager.activeSpan());
+
+ ContextManager.stopSpan(outer);
+ }
+
+ @Test
+ public void tracingFailureInsideStartExitSpanNeverBreaksTheRealRequest() throws Exception {
+ AbstractSpan outer = ContextManager.createLocalSpan("outer");
+ HttpRequest badRequest = mock(HttpRequest.class);
+ when(badRequest.getUri()).thenThrow(new java.net.URISyntaxException("x", "bad"));
+
+ RequestChannel realChannel = mock(RequestChannel.class);
+ AsyncRequestSpans spans = new AsyncRequestSpans(TARGET);
+ AsyncRequestProducerWrapper wrapper = new AsyncRequestProducerWrapper(syncDelegate(badRequest), spans);
+
+ wrapper.sendRequest(realChannel, mock(HttpContext.class));
+ verify(realChannel).sendRequest(eq(badRequest), any(), any(HttpContext.class));
+
+ ContextManager.stopSpan(outer);
+ }
+}
\ No newline at end of file
diff --git a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/AsyncResponseConsumerWrapperTest.java b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/AsyncResponseConsumerWrapperTest.java
new file mode 100644
index 0000000000..9b19816e2d
--- /dev/null
+++ b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/AsyncResponseConsumerWrapperTest.java
@@ -0,0 +1,199 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpclient.v5.wrapper;
+
+import java.util.Collections;
+import org.apache.hc.core5.concurrent.FutureCallback;
+import org.apache.hc.core5.http.EntityDetails;
+import org.apache.hc.core5.http.HttpResponse;
+import org.apache.hc.core5.http.nio.AsyncResponseConsumer;
+import org.apache.hc.core5.http.nio.CapacityChannel;
+import org.apache.hc.core5.http.protocol.HttpContext;
+import org.apache.hc.core5.http.protocol.HttpCoreContext;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
+import org.apache.skywalking.apm.plugin.httpclient.v5.AsyncRequestSpans;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * {@link AsyncResponseConsumerWrapper} never touches {@code ContextManager} — it only tags/finishes the
+ * {@link AsyncRequestSpans} reference it's given, so these tests wire a REAL {@code AsyncRequestSpans} to a
+ * MOCKED {@link AbstractSpan}, the same technique {@code AsyncRequestSpansTest} uses. That gives two things at
+ * once per test: proof the wrapped consumer's original behavior is still invoked unchanged (delegation), and
+ * proof of the actual span-lifecycle side effect through the real holder (not just "a mock was called").
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class AsyncResponseConsumerWrapperTest {
+
+ @Mock
+ private AbstractSpan span;
+
+ @Mock
+ private AsyncResponseConsumer