You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by wu...@apache.org on 2026/09/27 12:25:53 UTC
(skywalking-java) branch main updated: Fix HttpClient 5.x callback span handling (#832)
This is an automated email from the ASF dual-hosted git repository.
wu-sheng pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-java.git
The following commit(s) were added to refs/heads/main by this push:
new 1909516384 Fix HttpClient 5.x callback span handling (#832)
1909516384 is described below
commit 19095163842d199571c3e85e25b6e291915a9bd2
Author: Ayush Kulshreshtha <ay...@gmail.com>
AuthorDate: Sun Sep 27 17:55:48 2026 +0530
Fix HttpClient 5.x callback span handling (#832)
Fix HttpClient 5.x async span lifecycle and propagation
Create and detach async exit spans in the caller's context, then finish
them by reference without disturbing the caller or reactor thread's
span stack. Remove the IOSession poll instrumentation.
Preserve response-processing exceptions, handle cancellation and resource
release safely, and retain request authority and encoded URL paths.
Enable async tracing for HttpClient 5.4+ and expand regression tests and
scenario coverage across 5.0–5.6.
Fixes apache/skywalking#14097
---
CHANGES.md | 17 ++
.../plugin/httpclient/v5/AsyncRequestSpans.java | 167 +++++++++++
.../apm/plugin/httpclient/v5/Constants.java | 23 --
.../v5/HttpAsyncClientDoExecuteInterceptor.java | 81 ++++-
.../v5/IOSessionImplPollInterceptor.java | 92 ------
.../v5/define/IOSessionImplInstrumentation.java | 68 -----
.../v5/wrapper/AsyncRequestProducerWrapper.java | 179 +++++++++++
.../v5/wrapper/AsyncResponseConsumerWrapper.java | 110 ++++---
.../v5/wrapper/FutureCallbackWrapper.java | 46 +--
.../src/main/resources/skywalking-plugin.def | 1 -
.../httpclient/v5/AsyncRequestSpansTest.java | 188 ++++++++++++
.../wrapper/AsyncRequestProducerWrapperTest.java | 331 +++++++++++++++++++++
.../wrapper/AsyncResponseConsumerWrapperTest.java | 319 ++++++++++++++++++++
.../v5/wrapper/FutureCallbackWrapperTest.java | 129 ++++++++
.../service-agent/java-agent/Supported-list.md | 2 +-
.../config/expectedData.yaml | 21 +-
.../httpclient-5.x-scenario/support-version.list | 5 +
17 files changed, 1505 insertions(+), 274 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 93c6131ca5..add61965cb 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -12,6 +12,23 @@ Release Notes.
A concurrent request could overwrite it before the chain was subscribed, and the outbound `sw8` header then
carried a context the downstream service joined by mistake. The snapshot is held by a client derived per
request now (apache/skywalking#14095).
+* Fix the `httpclient-5.x-plugin` async client tracing (apache/skywalking#14097):
+ * `FutureCallback` no longer stops a span. With `HttpAsyncClients.classic(...)` the callback runs on the caller
+ thread, where it used to close the caller's own active span.
+ * The HTTP exit span is now created in the caller's context when the request is sent, detached right away, and
+ finished by reference when the response ends. Nothing is left on the I/O reactor thread's span stack any more,
+ so overlapping requests on one reactor thread no longer nest into or close each other's spans.
+ * The `httpasyncclient/local` span and its cross-thread segment are removed. The HTTP exit span now lives in the
+ caller's own segment.
+ * HttpClient 5.4+ async requests are traced now. Previously the plugin produced no spans for them, because
+ `doExecute` received a `null` `HttpContext` and the request was not yet in the context when `IOSessionImpl#poll`
+ ran.
+ * A response that fails after its head, while the body is read or when the consumer builds the result at EOF,
+ now ends the span as an error with the exception logged, instead of as a success or an error without a cause.
+ * The destination is taken from the explicit target or the request authority, as the client routes it, so
+ routable names that `java.net.URI` does not parse as a host (e.g. `service_name`) are traced and propagated.
+ * The `url` tag keeps the path encoded as it was sent (e.g. `%2F` stays `%2F`).
+ * The `httpclient-5.x-scenario` now tests one version per minor, 5.0 to 5.6.
* 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
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..e054b2baab
--- /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,167 @@
+/*
+ * 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.
+ *
+ * <p>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.
+ *
+ * <p>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<Thread> creator = new AtomicReference<>(Thread.currentThread());
+
+ private AbstractSpan span;
+ private boolean finished;
+
+ /**
+ * Set once the response head has been handed to the response consumer. From then on the client reports a
+ * failure through {@code failed(cause)}, see {@link #release()}.
+ */
+ private boolean responseStarted;
+
+ 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() {
+ return creator.compareAndSet(Thread.currentThread(), 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) {
+ responseStarted = true;
+
+ 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);
+ }
+
+ /**
+ * The exchange was cancelled before it 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);
+ }
+
+ /**
+ * The response consumer released its resources. In the normal case the span was already finished at
+ * {@code streamEnd}/{@code consumeResponse}, so this does nothing.
+ *
+ * <p>If the response head was already seen and the span is still open, the body failed:
+ * {@code HttpAsyncMainClientExec#failed} (and the H2 equivalent) releases the entity consumer first and then
+ * reports the cause through {@code failed(cause)}. Finishing here would drop that exception, so the span is
+ * left open for {@link #fail(Throwable)}.
+ *
+ * <p>If no response head was seen, release may be the only signal we get (e.g. a suppressed redirect with a
+ * non-repeatable entity in 5.5.x never calls {@code failed} or {@code completed}), so the span is ended as
+ * an error here.
+ */
+ public synchronized void release() {
+ if (responseStarted) {
+ return;
+ }
+
+ 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/Constants.java b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/Constants.java
deleted file mode 100644
index 2497ca8cbc..0000000000
--- a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpclient/v5/Constants.java
+++ /dev/null
@@ -1,23 +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;
-
-public class Constants {
-
- public static String SKYWALKING_CONTEXT_SNAPSHOT = "skywalking-context-snapshot";
-}
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..8736dbcd1c 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.
+ *
+ * <p>It does not store anything in the {@code HttpContext} and never touches the span stack of a thread other than
+ * the caller's. It only:
+ * <ol>
+ * <li>creates a per-request {@link AsyncRequestSpans} holder, while the caller's context is still active;</li>
+ * <li>wraps the request producer so the exit span is created on the caller thread, synchronously, the moment the
+ * concrete {@code HttpRequest} becomes available;</li>
+ * <li>wraps the response consumer and future callback so the retained span is finished by reference.</li>
+ * </ol>
+ * 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<MethodDescription> 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..a33415147e
--- /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,179 @@
+/*
+ * 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.hc.core5.net.URIAuthority;
+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 {
+ // Resolve the destination the way the client does (InternalAbstractHttpAsyncClient with
+ // RoutingSupport#determineHost): an explicit target wins, otherwise the request's scheme and authority.
+ // The authority is used as is, not re-parsed through java.net.URI, whose getHost() is null for routable
+ // names such as service_name.
+ HttpHost target = spans.getTarget();
+ String scheme;
+ String host;
+ int port;
+ if (target != null) {
+ scheme = target.getSchemeName();
+ host = target.getHostName();
+ port = target.getPort();
+ } else {
+ URIAuthority authority = request.getAuthority();
+ if (authority == null) {
+ return;
+ }
+ scheme = request.getScheme();
+ host = authority.getHostName();
+ port = authority.getPort();
+ }
+ if (host == null || host.isEmpty()) {
+ return;
+ }
+ if (scheme == null) {
+ scheme = "http";
+ }
+ if (port < 0) {
+ port = "https".equalsIgnoreCase(scheme) ? 443 : 80;
+ }
+ String peer = host + ":" + port;
+
+ URI uri = request.getUri();
+ String path = uri.getPath() == null || uri.getPath().isEmpty() ? "/" : uri.getPath();
+ // The URL tag keeps the path as it goes on the wire; the decoded path would turn %2F, %3F, %23 and %20
+ // into /, ?, # and a space and change what the URL means.
+ String rawPath = uri.getRawPath() == null || uri.getRawPath().isEmpty() ? "/" : uri.getRawPath();
+ String url = scheme + "://" + peer + rawPath + (uri.getRawQuery() == null ? "" : "?" + uri.getRawQuery());
+
+ // Inside another plugin's exit span, createExitSpan reuses that span (depth + 1) instead of creating one.
+ // That span belongs to the other plugin, so it must not be turned into an async span here: only propagate.
+ boolean nested = ContextManager.activeSpan().isExit();
+ // Create the span without a carrier and inject afterwards. createExitSpan(op, carrier, peer) injects before
+ // returning, and injection throws for a reused outer exit span without a peer, which would leave the extra
+ // depth on the caller's stack with nothing to stop it.
+ AbstractSpan span = ContextManager.createExitSpan(path, 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);
+ }
+ ContextCarrier carrier = new ContextCarrier();
+ ContextManager.inject(carrier);
+ CarrierItem next = carrier.items();
+ while (next.hasNext()) {
+ next = next.next();
+ request.setHeader(next.getHeadKey(), next.getHeadValue());
+ }
+ } finally {
+ // Detach before the request is forwarded: the client may report a failure on this thread before
+ // doExecute returns. prepareForAsync() requires the span to still be the active one.
+ if (!nested) {
+ span.prepareForAsync();
+ }
+ ContextManager.stopSpan(span);
+ if (!nested) {
+ spans.start(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();
+ }
+}
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..93a3d264d0 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,10 +13,14 @@
* 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.nio.ByteBuffer;
+import java.util.List;
import org.apache.hc.core5.concurrent.FutureCallback;
import org.apache.hc.core5.http.EntityDetails;
import org.apache.hc.core5.http.Header;
@@ -25,57 +29,54 @@ 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;
+/**
+ * Finishes the request's exit span through {@link AsyncRequestSpans}, never through the span stack of the current
+ * thread: the response is consumed on the I/O reactor thread, which serves many requests, and for
+ * {@code HttpAsyncClients.classic(...)} the body may be read to the end on the caller's thread.
+ */
public class AsyncResponseConsumerWrapper<T> implements AsyncResponseConsumer<T> {
- private AsyncResponseConsumer<T> consumer;
+ private final AsyncResponseConsumer<T> consumer;
+ private final AsyncRequestSpans spans;
- public AsyncResponseConsumerWrapper(AsyncResponseConsumer<T> consumer) {
+ public AsyncResponseConsumerWrapper(AsyncResponseConsumer<T> consumer, AsyncRequestSpans spans) {
this.consumer = consumer;
+ this.spans = spans;
}
@Override
public void consumeResponse(HttpResponse response, EntityDetails entityDetails, HttpContext context,
- FutureCallback<T> 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();
+ FutureCallback<T> resultCallback) throws HttpException, IOException {
+ spans.onResponse(response.getCode());
+ consumer.consumeResponse(response, entityDetails, context,
+ resultCallback == null ? null : new ResultCallback<>(resultCallback, spans));
+ if (entityDetails == null) {
+ // No body means streamEnd() will never be called for this exchange. Finish only once the consumer has
+ // processed the response: if it throws, the client reports the cause through failed(cause).
+ 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(List<? extends Header> trailers) throws HttpException, IOException {
+ // The consumer decodes the body and builds the result at EOF, and that can fail. Finish as a success only
+ // after it returned normally: a failure it throws reaches failed(cause), and a failure it reports through
+ // the result callback reaches ResultCallback#failed, both of which end the span with the cause.
+ consumer.streamEnd(trailers);
+ spans.finish();
+ }
+
@Override
public void failed(Exception cause) {
- if (ContextManager.isActive()) {
- ContextManager.activeSpan().errorOccurred().log(cause);
- ContextManager.stopSpan();
- }
+ spans.fail(cause);
consumer.failed(cause);
}
@@ -89,13 +90,48 @@ public class AsyncResponseConsumerWrapper<T> implements AsyncResponseConsumer<T>
consumer.consume(src);
}
- @Override
- public void streamEnd(List<? extends Header> trailers) throws HttpException, IOException {
- consumer.streamEnd(trailers);
- }
-
@Override
public void releaseResources() {
+ // Not a success signal. Normally the span was already finished at streamEnd/consumeResponse and this does
+ // nothing. HttpAsyncMainClientExec#failed and H2AsyncMainClientExec#failed release the consumer before
+ // they report the cause, so once the response head was seen the span stays open here and failed(cause)
+ // right after ends it with the exception logged. Only when no response head was seen does release end the
+ // span as an error, because it can be the only signal (a suppressed redirect with a non-repeatable entity
+ // in 5.5.x never calls failed or completed).
+ spans.release();
consumer.releaseResources();
}
+
+ /**
+ * The callback the consumer reports its result through. A failure there (e.g. an {@code EOFException} while
+ * building the response at {@code streamEnd}) ends the span with the cause. Success is left to
+ * {@code streamEnd}/{@code consumeResponse}, so the moment the span finishes does not change.
+ */
+ private static final class ResultCallback<T> implements FutureCallback<T> {
+
+ private final FutureCallback<T> callback;
+ private final AsyncRequestSpans spans;
+
+ ResultCallback(FutureCallback<T> callback, AsyncRequestSpans spans) {
+ this.callback = callback;
+ this.spans = spans;
+ }
+
+ @Override
+ public void completed(T result) {
+ callback.completed(result);
+ }
+
+ @Override
+ public void failed(Exception ex) {
+ spans.fail(ex);
+ callback.failed(ex);
+ }
+
+ @Override
+ public void cancelled() {
+ spans.abort();
+ callback.cancelled();
+ }
+ }
}
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 <em>whatever thread happens to
+ * invoke this callback</em> — 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.
+ *
+ * <p>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.
+ *
+ * <p>{@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<T> implements FutureCallback<T> {
- private FutureCallback<T> callback;
+ private final FutureCallback<T> callback;
+ private final AsyncRequestSpans spans;
- public FutureCallbackWrapper(FutureCallback<T> callback) {
+ public FutureCallbackWrapper(FutureCallback<T> 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..b42c9b7cbf
--- /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,188 @@
+/*
+ * 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.release(); // what releaseResources() calls; must be a no-op once already finished
+
+ verify(span, times(1)).asyncFinish();
+ verify(span, never()).errorOccurred();
+ }
+
+ @Test
+ public void releaseAfterResponseHeadWaitsForFailedAndKeepsTheCause() {
+ // HttpAsyncMainClientExec#failed releases the consumer BEFORE it reports the cause. With the response head
+ // already seen, release must leave the span open so the following fail(cause) still logs the exception.
+ RuntimeException cause = new RuntimeException("body read failed");
+ spans.onResponse(200);
+ spans.release();
+
+ verify(span, never()).asyncFinish();
+
+ spans.fail(cause);
+
+ verify(span, times(1)).errorOccurred();
+ verify(span, times(1)).log(cause);
+ verify(span, times(1)).asyncFinish();
+ }
+
+ @Test
+ public void releaseWithoutAnyResponseEndsAsError() {
+ // No response head (e.g. a suppressed redirect with a non-repeatable entity): release may be the only
+ // signal, so it ends the span as an error.
+ spans.release();
+
+ verify(span, times(1)).errorOccurred();
+ verify(span, never()).log(org.mockito.ArgumentMatchers.any(Throwable.class));
+ verify(span, times(1)).asyncFinish();
+ }
+
+ @Test
+ public void cancellationAfterResponseHeadStillEnds() {
+ spans.onResponse(200);
+ spans.release();
+ spans.abort();
+
+ 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..b25f7b243e
--- /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,331 @@
+/*
+ * 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.List;
+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.hc.core5.net.URIAuthority;
+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.core.context.trace.AbstractTracingSpan;
+import org.apache.skywalking.apm.agent.core.context.trace.TraceSegment;
+import org.apache.skywalking.apm.agent.core.context.util.TagValuePair;
+import org.apache.skywalking.apm.agent.test.helper.SegmentHelper;
+import org.apache.skywalking.apm.agent.test.helper.SpanHelper;
+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.assertFalse;
+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;
+
+/**
+ * Runs {@link AsyncRequestProducerWrapper} against the real {@link ContextManager}: the exit span must be created in
+ * the caller's segment, detached from the caller's stack before the request is forwarded, and finished later by
+ * reference only.
+ */
+@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);
+ }
+
+ @Test
+ public void exitSpanLandsInCallerSegmentWithExplicitTargetAsPeer() throws Exception {
+ AbstractSpan caller = ContextManager.createLocalSpan("caller");
+ // The request's own authority differs from the explicit target, which must win, as in the client itself.
+ AsyncRequestSpans spans = new AsyncRequestSpans(TARGET);
+ AsyncRequestProducerWrapper wrapper = new AsyncRequestProducerWrapper(
+ syncDelegate(requestTo("http://other.invalid:9999/hello?a=b")), spans);
+
+ wrapper.sendRequest(mock(RequestChannel.class), mock(HttpContext.class));
+ spans.onResponse(200);
+ spans.finish();
+ ContextManager.stopSpan(caller);
+
+ assertEquals(1, segmentStorage.getTraceSegments().size());
+ TraceSegment segment = segmentStorage.getTraceSegments().get(0);
+ List<AbstractTracingSpan> spanList = SegmentHelper.getSpans(segment);
+ assertEquals(2, spanList.size());
+ AbstractTracingSpan exit = spanList.get(0);
+ assertTrue(exit.isExit());
+ assertEquals("/hello", exit.getOperationName());
+ assertEquals("example.org:8080", SpanHelper.getPeer(exit));
+ assertEquals(caller.getSpanId(), SpanHelper.getParentSpanId(exit));
+ assertFalse(SpanHelper.getErrorOccurred(exit));
+ }
+
+ /**
+ * Without an explicit target the client routes by the request's scheme and authority. {@code service_name} is
+ * routable (e.g. through a custom DNS resolver) but java.net.URI does not parse it as a host, so the destination
+ * must come from the authority, not from {@code URI.getHost()}.
+ */
+ @Test
+ public void routableAuthorityThatUriCannotParseIsStillTraced() throws Exception {
+ AbstractSpan caller = ContextManager.createLocalSpan("caller");
+ HttpRequest request = new BasicHttpRequest("GET", "/hello?a=b");
+ request.setScheme("http");
+ request.setAuthority(new URIAuthority("service_name", 8080));
+ AsyncRequestSpans spans = new AsyncRequestSpans(null);
+ 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"));
+ spans.onResponse(200);
+ spans.finish();
+ ContextManager.stopSpan(caller);
+
+ AbstractTracingSpan exit = SegmentHelper.getSpans(segmentStorage.getTraceSegments().get(0)).get(0);
+ assertTrue(exit.isExit());
+ assertEquals("/hello", exit.getOperationName());
+ assertEquals("service_name:8080", SpanHelper.getPeer(exit));
+ assertEquals("http://service_name:8080/hello?a=b", tag(exit, "url"));
+ }
+
+ @Test
+ public void noTargetAndNoAuthorityCreatesNoSpan() throws Exception {
+ AbstractSpan caller = ContextManager.createLocalSpan("caller");
+ HttpRequest request = new BasicHttpRequest("GET", "/hello");
+ RequestChannel realChannel = mock(RequestChannel.class);
+ AsyncRequestSpans spans = new AsyncRequestSpans(null);
+ AsyncRequestProducerWrapper wrapper = new AsyncRequestProducerWrapper(syncDelegate(request), spans);
+
+ wrapper.sendRequest(realChannel, mock(HttpContext.class));
+
+ verify(realChannel).sendRequest(eq(request), any(), any(HttpContext.class));
+ assertFalse(request.containsHeader("sw8"));
+ assertSame(caller, ContextManager.activeSpan());
+ ContextManager.stopSpan(caller);
+ }
+
+ /**
+ * The URL tag must keep the path encoded as it is sent; decoding it would turn %2F, %3F, %23 and %20 into
+ * characters that change what the URL means. The operation name keeps using the decoded path.
+ */
+ @Test
+ public void urlTagKeepsTheEncodedPath() throws Exception {
+ AbstractSpan caller = ContextManager.createLocalSpan("caller");
+ AsyncRequestSpans spans = new AsyncRequestSpans(TARGET);
+ AsyncRequestProducerWrapper wrapper = new AsyncRequestProducerWrapper(
+ syncDelegate(requestTo("http://example.org:8080/a%2Fb%3Fc%23d%20e?q=value")), spans);
+
+ wrapper.sendRequest(mock(RequestChannel.class), mock(HttpContext.class));
+ spans.finish();
+ ContextManager.stopSpan(caller);
+
+ AbstractTracingSpan exit = SegmentHelper.getSpans(segmentStorage.getTraceSegments().get(0)).get(0);
+ assertEquals("http://example.org:8080/a%2Fb%3Fc%23d%20e?q=value", tag(exit, "url"));
+ assertEquals("/a/b?c#d e", exit.getOperationName());
+ }
+
+ private static String tag(AbstractTracingSpan span, String key) {
+ for (TagValuePair pair : SpanHelper.getTags(span)) {
+ if (key.equals(pair.getKey().key())) {
+ return pair.getValue();
+ }
+ }
+ return null;
+ }
+
+ /**
+ * An outer exit span without a peer: injection fails, but the reused span's extra depth must still be released,
+ * otherwise the outer span could never be stopped and the caller's segment would never be reported.
+ */
+ @Test
+ public void nestedInsideExitSpanWithoutPeerLeavesTheStackBalanced() throws Exception {
+ AbstractSpan outerExit = ContextManager.createExitSpan("outer-exit", "");
+ HttpRequest request = requestTo("http://example.org/hello");
+ RequestChannel realChannel = mock(RequestChannel.class);
+ AsyncRequestSpans spans = new AsyncRequestSpans(TARGET);
+ AsyncRequestProducerWrapper wrapper = new AsyncRequestProducerWrapper(syncDelegate(request), spans);
+
+ wrapper.sendRequest(realChannel, mock(HttpContext.class));
+
+ verify(realChannel).sendRequest(eq(request), any(), any(HttpContext.class));
+ assertSame(outerExit, ContextManager.activeSpan());
+ ContextManager.stopSpan(outerExit);
+ assertEquals(1, segmentStorage.getTraceSegments().size());
+ }
+}
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..2a4268c06b
--- /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,319 @@
+/*
+ * 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.EOFException;
+import java.io.IOException;
+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.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.fail;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doThrow;
+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<Object> delegate;
+
+ @Mock
+ private FutureCallback<Object> resultCallback;
+
+ private AsyncRequestSpans spans;
+ private AsyncResponseConsumerWrapper<Object> wrapper;
+ private HttpContext context;
+
+ @Before
+ public void setUp() {
+ spans = new AsyncRequestSpans(null);
+ spans.start(span);
+ wrapper = new AsyncResponseConsumerWrapper<>(delegate, spans);
+ context = new HttpCoreContext();
+ }
+
+ private HttpResponse response(int status) {
+ HttpResponse r = mock(HttpResponse.class);
+ when(r.getCode()).thenReturn(status);
+ return r;
+ }
+
+ @Test
+ public void consumeResponseWithEntityTagsStatusButDoesNotFinishYet() throws Exception {
+ EntityDetails entity = mock(EntityDetails.class);
+ HttpResponse response = response(200);
+
+ wrapper.consumeResponse(response, entity, context, resultCallback);
+
+ verify(span, never()).asyncFinish();
+ verify(delegate).consumeResponse(eq(response), eq(entity), eq(context), any());
+ }
+
+ @Test
+ public void consumeResponseWithoutEntityFinishesImmediately() throws Exception {
+ // e.g. a 204 with no body: streamEnd() will never be called for this exchange, so consumeResponse()
+ // itself must finish the span.
+ HttpResponse response = response(204);
+
+ wrapper.consumeResponse(response, null, context, resultCallback);
+
+ verify(span, times(1)).asyncFinish();
+ verify(delegate).consumeResponse(eq(response), isNull(), eq(context), any());
+ }
+
+ @Test
+ public void consumeResponseWithoutEntityFailingIsNotReportedAsSuccess() throws Exception {
+ HttpResponse response = response(204);
+ IOException cause = new IOException("consumer rejected the response");
+ doThrow(cause).when(delegate).consumeResponse(eq(response), isNull(), eq(context), any());
+
+ try {
+ wrapper.consumeResponse(response, null, context, resultCallback);
+ fail("the consumer's exception must propagate to the client");
+ } catch (IOException expected) {
+ assertSame(cause, expected);
+ }
+ verify(span, never()).asyncFinish();
+
+ wrapper.releaseResources();
+ wrapper.failed(cause);
+
+ verify(span, times(1)).errorOccurred();
+ verify(span, times(1)).log(cause);
+ verify(span, times(1)).asyncFinish();
+ }
+
+ @Test
+ public void streamEndFailureIsNotReportedAsSuccess() throws Exception {
+ // The consumer throws while building the response at EOF. The span must not be finished as a success
+ // first, so the failed(cause) the client reports afterwards still records the error and the exception.
+ EOFException cause = new EOFException("truncated body");
+ doThrow(cause).when(delegate).streamEnd(any());
+ wrapper.consumeResponse(response(200), mock(EntityDetails.class), context, resultCallback);
+
+ try {
+ wrapper.streamEnd(Collections.emptyList());
+ fail("the consumer's exception must propagate to the client");
+ } catch (EOFException expected) {
+ assertSame(cause, expected);
+ }
+ verify(span, never()).asyncFinish();
+
+ wrapper.releaseResources();
+ wrapper.failed(cause);
+
+ verify(span, times(1)).errorOccurred();
+ verify(span, times(1)).log(cause);
+ verify(span, times(1)).asyncFinish();
+ }
+
+ @Test
+ public void streamEndFailureReportedThroughResultCallbackIsAnError() throws Exception {
+ // The consumer reports the EOF failure through the result callback it was given and returns normally.
+ EOFException cause = new EOFException("truncated body");
+ ArgumentCaptor<FutureCallback<Object>> given = ArgumentCaptor.forClass(FutureCallback.class);
+ EntityDetails entity = mock(EntityDetails.class);
+ HttpResponse response = response(200);
+ wrapper.consumeResponse(response, entity, context, resultCallback);
+ verify(delegate).consumeResponse(eq(response), eq(entity), eq(context), given.capture());
+ doAnswer(invocation -> {
+ given.getValue().failed(cause);
+ return null;
+ }).when(delegate).streamEnd(any());
+
+ wrapper.streamEnd(Collections.emptyList());
+
+ verify(span, times(1)).errorOccurred();
+ verify(span, times(1)).log(cause);
+ verify(span, times(1)).asyncFinish();
+ verify(resultCallback).failed(cause);
+ }
+
+ @Test
+ public void resultCallbackSuccessIsPassedThroughAndFinishesAtStreamEnd() throws Exception {
+ ArgumentCaptor<FutureCallback<Object>> given = ArgumentCaptor.forClass(FutureCallback.class);
+ EntityDetails entity = mock(EntityDetails.class);
+ HttpResponse response = response(200);
+ Object result = new Object();
+ wrapper.consumeResponse(response, entity, context, resultCallback);
+ verify(delegate).consumeResponse(eq(response), eq(entity), eq(context), given.capture());
+ doAnswer(invocation -> {
+ given.getValue().completed(result);
+ return null;
+ }).when(delegate).streamEnd(any());
+
+ wrapper.streamEnd(Collections.emptyList());
+
+ verify(resultCallback).completed(result);
+ verify(span, never()).errorOccurred();
+ verify(span, times(1)).asyncFinish();
+ }
+
+ @Test
+ public void errorStatusMarksErrorWithoutFinishing() throws Exception {
+ EntityDetails entity = mock(EntityDetails.class);
+ HttpResponse response = response(500);
+
+ wrapper.consumeResponse(response, entity, context, resultCallback);
+
+ verify(span, times(1)).errorOccurred();
+ verify(span, never()).asyncFinish();
+ }
+
+ @Test
+ public void informationResponseNeverTouchesTheSpan() throws Exception {
+ HttpResponse response = response(100);
+
+ wrapper.informationResponse(response, context);
+
+ verify(span, never()).asyncFinish();
+ verify(span, never()).errorOccurred();
+ verify(delegate).informationResponse(response, context);
+ }
+
+ @Test
+ public void streamEndFinishesTheSpanExactlyOnce() throws Exception {
+ wrapper.streamEnd(Collections.emptyList());
+
+ verify(span, times(1)).asyncFinish();
+ verify(delegate).streamEnd(Collections.emptyList());
+ }
+
+ @Test
+ public void consumeResponseThenStreamEndFinishesExactlyOnce() throws Exception {
+ EntityDetails entity = mock(EntityDetails.class);
+ wrapper.consumeResponse(response(200), entity, context, resultCallback);
+ wrapper.streamEnd(Collections.emptyList());
+
+ verify(span, times(1)).asyncFinish();
+ }
+
+ @Test
+ public void failedMarksErrorAndFinishesExactlyOnce() {
+ RuntimeException cause = new RuntimeException("boom");
+
+ wrapper.failed(cause);
+
+ verify(span, times(1)).errorOccurred();
+ verify(span, times(1)).log(cause);
+ verify(span, times(1)).asyncFinish();
+ verify(delegate).failed(cause);
+ }
+
+ @Test
+ public void releaseResourcesAfterNormalCompletionIsANoOp() throws Exception {
+ wrapper.streamEnd(Collections.emptyList());
+ wrapper.releaseResources();
+
+ // finish() already ran at streamEnd(); releaseResources() must not run it a second time nor
+ // retroactively mark a successful exchange as an error.
+ verify(span, times(1)).asyncFinish();
+ verify(span, never()).errorOccurred();
+ verify(delegate).releaseResources();
+ }
+
+ @Test
+ public void bodyFailureAfterResponseHeadKeepsStatusErrorAndException() throws Exception {
+ // HttpAsyncMainClientExec#failed / H2AsyncMainClientExec#failed release the consumer BEFORE they report
+ // the cause: onResponse(200) -> releaseResources() -> failed(cause). The span must end once, as an error,
+ // with the exception logged, instead of being finished without a cause by the release.
+ RuntimeException cause = new RuntimeException("connection reset while reading the body");
+ EntityDetails entity = mock(EntityDetails.class);
+
+ wrapper.consumeResponse(response(200), entity, context, resultCallback);
+ wrapper.releaseResources();
+
+ verify(span, never()).asyncFinish();
+ verify(delegate).releaseResources();
+
+ wrapper.failed(cause);
+
+ verify(span, times(1)).errorOccurred();
+ verify(span, times(1)).log(cause);
+ verify(span, times(1)).asyncFinish();
+ verify(delegate).failed(cause);
+ }
+
+ @Test
+ public void releaseResourcesWithoutResponseThenFailedEndsExactlyOnce() {
+ // No response head: release ends the span as an error, and a later failed() must not end it again.
+ wrapper.releaseResources();
+ wrapper.failed(new RuntimeException("arrives after release"));
+
+ verify(span, times(1)).errorOccurred();
+ verify(span, times(1)).asyncFinish();
+ }
+
+ @Test
+ public void releaseResourcesWithoutAnyResponseEndsAsError() {
+ // A suppressed redirect with a non-repeatable entity: only releaseResources() ever runs, failed()/
+ // completed() never do. The span must still end, and must end as an error (the exchange never actually
+ // completed), not silently disappear.
+ wrapper.releaseResources();
+
+ verify(span, times(1)).errorOccurred();
+ verify(span, times(1)).asyncFinish();
+ }
+
+ @Test
+ public void updateCapacityAndConsumeAreTransparentPassthroughs() throws Exception {
+ CapacityChannel capacityChannel = mock(CapacityChannel.class);
+ wrapper.updateCapacity(capacityChannel);
+ verify(delegate).updateCapacity(capacityChannel);
+ verify(span, never()).asyncFinish();
+
+ java.nio.ByteBuffer buf = java.nio.ByteBuffer.allocate(0);
+ wrapper.consume(buf);
+ verify(delegate).consume(buf);
+ verify(span, never()).asyncFinish();
+ }
+}
diff --git a/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/FutureCallbackWrapperTest.java b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/FutureCallbackWrapperTest.java
new file mode 100644
index 0000000000..70c0998072
--- /dev/null
+++ b/apm-sniffer/apm-sdk-plugin/httpclient-5.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/httpclient/v5/wrapper/FutureCallbackWrapperTest.java
@@ -0,0 +1,129 @@
+/*
+ * 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 org.apache.hc.core5.concurrent.FutureCallback;
+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.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+/**
+ * This is the class the original bug (#14097) lived in. The old behavior called the parameterless
+ * {@code ContextManager.stopSpan()} here, which is exactly what these tests exist to guard against ever
+ * regressing to: none of them touch {@code ContextManager} at all, they only assert the {@link AsyncRequestSpans}
+ * reference is finished by reference — which is what actually makes it safe to run on the caller/business
+ * thread, as {@code HttpAsyncClients.classic(...)} does.
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class FutureCallbackWrapperTest {
+
+ @Mock
+ private AbstractSpan span;
+
+ @Mock
+ private FutureCallback<String> delegate;
+
+ private AsyncRequestSpans spans;
+
+ @Before
+ public void setUp() {
+ spans = new AsyncRequestSpans(null);
+ spans.start(span);
+ }
+
+ @Test
+ public void completedFinishesSpanExactlyOnceAndDelegates() {
+ FutureCallbackWrapper<String> wrapper = new FutureCallbackWrapper<>(delegate, spans);
+
+ wrapper.completed("result");
+
+ verify(span, times(1)).asyncFinish();
+ verify(span, never()).errorOccurred();
+ verify(delegate).completed("result");
+ }
+
+ @Test
+ public void failedMarksErrorAndFinishesExactlyOnceAndDelegates() {
+ FutureCallbackWrapper<String> wrapper = new FutureCallbackWrapper<>(delegate, spans);
+ RuntimeException cause = new RuntimeException("boom");
+
+ wrapper.failed(cause);
+
+ verify(span, times(1)).errorOccurred();
+ verify(span, times(1)).log(cause);
+ verify(span, times(1)).asyncFinish();
+ verify(delegate).failed(cause);
+ }
+
+ @Test
+ public void cancelledMarksErrorAndFinishesExactlyOnceAndDelegates() {
+ FutureCallbackWrapper<String> wrapper = new FutureCallbackWrapper<>(delegate, spans);
+
+ wrapper.cancelled();
+
+ verify(span, times(1)).errorOccurred();
+ verify(span, times(1)).asyncFinish();
+ verify(delegate).cancelled();
+ }
+
+ @Test
+ public void toleratesANullDelegateCallback() {
+ // doExecute is always wrapped even when the caller passed no callback of their own — this is the only
+ // lifecycle hook that observes cancellation, so it must not NPE on a null delegate.
+ FutureCallbackWrapper<String> wrapper = new FutureCallbackWrapper<>(null, spans);
+
+ wrapper.completed("result"); // must not throw
+
+ verify(span, times(1)).asyncFinish();
+ }
+
+ @Test
+ public void completedAfterConsumerAlreadyFinishedDoesNotDoubleFinish() {
+ // Simulates AsyncResponseConsumerWrapper already having finished the span (streamEnd/consumeResponse)
+ // before the future callback also fires — FutureCallbackWrapper's own paths are largely redundant
+ // safety nets, and AsyncRequestSpans' idempotency is what makes that redundancy safe.
+ spans.finish(); // as if AsyncResponseConsumerWrapper already ran
+ FutureCallbackWrapper<String> wrapper = new FutureCallbackWrapper<>(delegate, spans);
+
+ wrapper.completed("result");
+
+ verify(span, times(1)).asyncFinish();
+ verify(delegate).completed("result");
+ }
+
+ @Test
+ public void cancelledAfterAlreadyFailedDoesNotOverwriteOrDoubleFinish() {
+ FutureCallbackWrapper<String> wrapper = new FutureCallbackWrapper<>(delegate, spans);
+ wrapper.failed(new RuntimeException("first"));
+
+ wrapper.cancelled();
+
+ verify(span, times(1)).errorOccurred();
+ verify(span, times(1)).asyncFinish();
+ verify(delegate).cancelled();
+ }
+}
diff --git a/docs/en/setup/service-agent/java-agent/Supported-list.md b/docs/en/setup/service-agent/java-agent/Supported-list.md
index 6e29b2fd24..42ff1dbac5 100644
--- a/docs/en/setup/service-agent/java-agent/Supported-list.md
+++ b/docs/en/setup/service-agent/java-agent/Supported-list.md
@@ -31,7 +31,7 @@ metrics based on the tracing data.
* [Feign](https://github.com/OpenFeign/feign) 9.x -> 13.5
* [Netflix Spring Cloud Feign](https://github.com/spring-cloud/spring-cloud-openfeign) 1.1.x -> 2.x
* [Okhttp](https://github.com/square/okhttp) 2.x -> 3.x -> 4.x
- * [Apache httpcomponent HttpClient](http://hc.apache.org/) 2.0 -> 3.1, 4.2, 4.3, 5.0, 5.1
+ * [Apache httpcomponent HttpClient](http://hc.apache.org/) 2.0 -> 3.1, 4.2, 4.3, 5.0 -> 5.6
* [Spring RestTemplate](https://github.com/spring-projects/spring-framework) 4.x
* [Spring RestTemplate](https://github.com/spring-projects/spring-framework) 6.x -> 7.x (Optional²)
* [Jetty Client](http://www.eclipse.org/jetty/) 9.x -> 11.x
diff --git a/test/plugin/scenarios/httpclient-5.x-scenario/config/expectedData.yaml b/test/plugin/scenarios/httpclient-5.x-scenario/config/expectedData.yaml
index d1d6e1d04f..bb3c8d7a99 100644
--- a/test/plugin/scenarios/httpclient-5.x-scenario/config/expectedData.yaml
+++ b/test/plugin/scenarios/httpclient-5.x-scenario/config/expectedData.yaml
@@ -15,7 +15,7 @@
# limitations under the License.
segmentItems:
- serviceName: httpclient-5.x-scenario
- segmentSize: ge 4
+ segmentSize: ge 3
segments:
- segmentId: not null
spans:
@@ -34,7 +34,7 @@ segmentItems:
- {key: http.method, value: GET}
- {key: http.status_code, value: '200'}
refs:
- - {parentEndpoint: httpasyncclient/local, networkAddress: '127.0.0.1:8080',
+ - {parentEndpoint: 'GET:/httpclient-5.x/case/asyncGet', networkAddress: '127.0.0.1:8080',
refType: CrossProcess, parentSpanId: 1, parentTraceSegmentId: not null, parentServiceInstance: not
null, parentService: httpclient-5.x-scenario, traceId: not null}
skipAnalysis: 'false'
@@ -55,23 +55,6 @@ segmentItems:
- {key: http.method, value: GET}
- {key: http.status_code, value: '200'}
skipAnalysis: 'false'
- - operationName: httpasyncclient/local
- parentSpanId: -1
- spanId: 0
- spanLayer: Http
- startTime: nq 0
- endTime: nq 0
- componentId: 26
- isError: false
- spanType: Local
- peer: ''
- refs:
- - {parentEndpoint: GET:/httpclient-5.x/case/asyncGet, networkAddress: '',
- refType: CrossThread, parentSpanId: 0, parentTraceSegmentId: not null, parentServiceInstance: not
- null, parentService: httpclient-5.x-scenario, traceId: not null}
- skipAnalysis: 'false'
- - segmentId: not null
- spans:
- operationName: GET:/httpclient-5.x/case/asyncGet
parentSpanId: -1
spanId: 0
diff --git a/test/plugin/scenarios/httpclient-5.x-scenario/support-version.list b/test/plugin/scenarios/httpclient-5.x-scenario/support-version.list
index ce0644ff9f..b6b73792e2 100644
--- a/test/plugin/scenarios/httpclient-5.x-scenario/support-version.list
+++ b/test/plugin/scenarios/httpclient-5.x-scenario/support-version.list
@@ -16,3 +16,8 @@
5.0
5.1
+5.2
+5.3
+5.4
+5.5
+5.6