Fix Jetty client cannot receive the HTTP response body(#13500)

CHANGES.md

@@ -16,6 +16,7 @@ Release Notes.
 * Update Maven to 3.6.3 in mvnw.
 * Fix OOM due to too many span logs.
 * Fix ClassLoader cache OOM issue with WeakHashMap.
+* Fix Jetty client cannot receive the HTTP response body.
 
 All issues and pull requests are [here](https://github.com/apache/skywalking/milestone/242?closed=1)
 

apm-sniffer/apm-sdk-plugin/jetty-plugin/jetty-client-9.0-plugin/src/main/java/org/apache/skywalking/apm/plugin/jetty/v90/client/AsyncHttpRequestSendInterceptor.java

@@ -58,8 +58,13 @@ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allAr
 
         span.prepareForAsync();
         request.attribute(Constants.SW_JETTY_EXIT_SPAN_KEY, span);
-        Response.CompleteListener callback = (Response.CompleteListener) allArguments[0];
-        allArguments[0] = new CompleteListenerWrapper(callback, ContextManager.capture());
+        if (allArguments[0] instanceof Response.Listener) {
+            Response.Listener listener = (Response.Listener) allArguments[0];
+            allArguments[0] = new ResponseListenerWrapper(listener, ContextManager.capture());
+        } else {
+            Response.CompleteListener listener = (Response.CompleteListener) allArguments[0];
+            allArguments[0] = new CompleteListenerWrapper(listener, ContextManager.capture());
+        }
     }
 
     @Override

apm-sniffer/apm-sdk-plugin/jetty-plugin/jetty-client-9.0-plugin/src/main/java/org/apache/skywalking/apm/plugin/jetty/v90/client/CompleteListenerWrapper.java

@@ -27,11 +27,11 @@
 import org.eclipse.jetty.client.api.Result;
 
 public class CompleteListenerWrapper implements Response.CompleteListener  {
-    private Response.CompleteListener callback;
+    private Response.CompleteListener listener;
     private ContextSnapshot context;
 
-    public CompleteListenerWrapper(Response.CompleteListener callback, ContextSnapshot context) {
-        this.callback = callback;
+    public CompleteListenerWrapper(Response.CompleteListener listener, ContextSnapshot context) {
+        this.listener = listener;
         this.context = context;
     }
 
@@ -43,9 +43,9 @@ public void onComplete(Result result) {
         if (context != null) {
             ContextManager.continued(context);
         }
-        if (callback != null) {
-            callback.onComplete(result);
+        if (listener != null) {
+            listener.onComplete(result);
         }
         ContextManager.stopSpan();
     }
-}
+}

apm-sniffer/apm-sdk-plugin/jetty-plugin/jetty-client-9.0-plugin/src/main/java/org/apache/skywalking/apm/plugin/jetty/v90/client/ResponseListenerWrapper.java

@@ -0,0 +1,85 @@
+/*
+ * 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.jetty.v90.client;
+
+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.trace.AbstractSpan;
+import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
+import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
+import org.eclipse.jetty.client.api.Response;
+import org.eclipse.jetty.client.api.Result;
+import org.eclipse.jetty.http.HttpField;
+import java.nio.ByteBuffer;
+
+public class ResponseListenerWrapper implements Response.Listener {
+
+    private final Response.Listener listener;
+
+    private final ContextSnapshot context;
+
+    public ResponseListenerWrapper(Response.Listener listener, ContextSnapshot context) {
+        this.listener = listener;
+        this.context = context;
+    }
+
+    @Override
+    public void onComplete(Result result) {
+        AbstractSpan span = ContextManager.createLocalSpan(Constants.PLUGIN_NAME + "/CompleteListener/onComplete");
+        span.setComponent(ComponentsDefine.JETTY_CLIENT);
+        SpanLayer.asHttp(span);
+        if (context != null) {
+            ContextManager.continued(context);
+        }
+        if (listener != null) {
+            listener.onComplete(result);
+        }
+        ContextManager.stopSpan();
+    }
+
+    @Override
+    public void onHeaders(Response response) {
+        listener.onHeaders(response);
+    }
+
+    @Override
+    public void onContent(Response response, ByteBuffer content) {
+        listener.onContent(response, content);
+    }
+
+    @Override
+    public void onBegin(Response response) {
+        listener.onBegin(response);
+    }
+
+    @Override
+    public boolean onHeader(Response response, HttpField field) {
+        return listener.onHeader(response, field);
+    }
+
+    @Override
+    public void onSuccess(Response response) {
+        listener.onSuccess(response);
+    }
+
+    @Override
+    public void onFailure(Response response, Throwable failure) {
+        listener.onFailure(response, failure);
+    }
+}

apm-sniffer/apm-sdk-plugin/jetty-plugin/jetty-client-9.0-plugin/src/main/java/org/apache/skywalking/apm/plugin/jetty/v90/client/define/HttpRequestInstrumentation.java

@@ -41,7 +41,7 @@ public class HttpRequestInstrumentation extends ClassInstanceMethodsEnhancePlugi
     private static final String ENHANCE_CLASS = "org.eclipse.jetty.client.HttpRequest";
     private static final String ENHANCE_CLASS_NAME = "send";
     public static final String SYNC_SEND_INTERCEPTOR =
-            "org.apache.skywalking.apm.plugin.jetty.v90.client.SyncHttpRequestSendV90Interceptor";
+            "org.apache.skywalking.apm.plugin.jetty.v90.client.SyncHttpRequestSendInterceptor";
 
     public static final String ASYNC_SEND_INTERCEPTOR =
             "org.apache.skywalking.apm.plugin.jetty.v90.client.AsyncHttpRequestSendInterceptor";
@@ -85,7 +85,7 @@ public String getMethodsInterceptor() {
 
                 @Override
                 public boolean isOverrideArgs() {
-                    return false;
+                    return true;
                 }
             }
         };

apm-sniffer/apm-sdk-plugin/jetty-plugin/jetty-client-9.0-plugin/src/main/java/org/apache/skywalking/apm/plugin/jetty/v90/client/define/ResponseNotifierInstrumentation.java

@@ -41,10 +41,10 @@
  */
 public class ResponseNotifierInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {
 
-    private static final String ENHANCE_CLASS = "org.eclipse.jetty.client.tar";
+    private static final String ENHANCE_CLASS = "org.eclipse.jetty.client.ResponseNotifier";
     private static final String ENHANCE_CLASS_NAME = "notifyComplete";
     public static final String SYNC_SEND_INTERCEPTOR =
-            "org.apache.skywalking.apm.plugin.jetty.v9.client.ResponseNotifierInterceptor";
+            "org.apache.skywalking.apm.plugin.jetty.v90.client.ResponseNotifierInterceptor";
 
     @Override
     public ConstructorInterceptPoint[] getConstructorsInterceptPoints() {

apm-sniffer/apm-sdk-plugin/jetty-plugin/jetty-client-9.x-plugin/pom.xml

@@ -31,7 +31,7 @@
     <url>http://maven.apache.org</url>
 
     <properties>
-        <jetty-client.version>9.1.0.v20131115</jetty-client.version>
+        <jetty-client.version>9.2.23.v20171218</jetty-client.version>
     </properties>
 
     <dependencies>

apm-sniffer/apm-sdk-plugin/jetty-plugin/jetty-client-9.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/jetty/v9/client/AsyncHttpRequestSendInterceptor.java

@@ -57,8 +57,13 @@ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allAr
 
         span.prepareForAsync();
         request.attribute(Constants.SW_JETTY_EXIT_SPAN_KEY, span);
-        Response.CompleteListener callback = (Response.CompleteListener) allArguments[0];
-        allArguments[0] = new CompleteListenerWrapper(callback, ContextManager.capture());
+        if (allArguments[0] instanceof Response.Listener) {
+            Response.Listener listener = (Response.Listener) allArguments[0];
+            allArguments[0] = new ResponseListenerWrapper(listener, ContextManager.capture());
+        } else {
+            Response.CompleteListener listener = (Response.CompleteListener) allArguments[0];
+            allArguments[0] = new CompleteListenerWrapper(listener, ContextManager.capture());
+        }
     }
 
     @Override

apm-sniffer/apm-sdk-plugin/jetty-plugin/jetty-client-9.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/jetty/v9/client/CompleteListenerWrapper.java

@@ -27,11 +27,11 @@
 import org.eclipse.jetty.client.api.Result;
 
 public class CompleteListenerWrapper implements Response.CompleteListener  {
-    private Response.CompleteListener callback;
+    private Response.CompleteListener listener;
     private ContextSnapshot context;
 
-    public CompleteListenerWrapper(Response.CompleteListener callback, ContextSnapshot context) {
-        this.callback = callback;
+    public CompleteListenerWrapper(Response.CompleteListener listener, ContextSnapshot context) {
+        this.listener = listener;
         this.context = context;
     }
 
@@ -43,9 +43,9 @@ public void onComplete(Result result) {
         if (context != null) {
             ContextManager.continued(context);
         }
-        if (callback != null) {
-            callback.onComplete(result);
+        if (listener != null) {
+            listener.onComplete(result);
         }
         ContextManager.stopSpan();
     }
-}
+}

apm-sniffer/apm-sdk-plugin/jetty-plugin/jetty-client-9.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/jetty/v9/client/ResponseListenerWrapper.java

@@ -0,0 +1,92 @@
+/*
+ * 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.jetty.v9.client;
+
+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.trace.AbstractSpan;
+import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
+import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
+import org.eclipse.jetty.client.api.Response;
+import org.eclipse.jetty.client.api.Result;
+import org.eclipse.jetty.http.HttpField;
+import org.eclipse.jetty.util.Callback;
+
+import java.nio.ByteBuffer;
+
+public class ResponseListenerWrapper implements Response.Listener {
+
+    private final Response.Listener listener;
+
+    private final ContextSnapshot context;
+
+    public ResponseListenerWrapper(Response.Listener listener, ContextSnapshot context) {
+        this.listener = listener;
+        this.context = context;
+    }
+
+    @Override
+    public void onComplete(Result result) {
+        AbstractSpan span = ContextManager.createLocalSpan(Constants.PLUGIN_NAME + "/CompleteListener/onComplete");
+        span.setComponent(ComponentsDefine.JETTY_CLIENT);
+        SpanLayer.asHttp(span);
+        if (context != null) {
+            ContextManager.continued(context);
+        }
+        if (listener != null) {
+            listener.onComplete(result);
+        }
+        ContextManager.stopSpan();
+    }
+
+    @Override
+    public void onHeaders(Response response) {
+        listener.onHeaders(response);
+    }
+
+    @Override
+    public void onContent(Response response, ByteBuffer content, Callback callback) {
+        listener.onContent(response, content, callback);
+    }
+
+    @Override
+    public void onContent(Response response, ByteBuffer content) {
+        listener.onContent(response, content);
+    }
+
+    @Override
+    public void onBegin(Response response) {
+        listener.onBegin(response);
+    }
+
+    @Override
+    public boolean onHeader(Response response, HttpField field) {
+        return listener.onHeader(response, field);
+    }
+
+    @Override
+    public void onSuccess(Response response) {
+        listener.onSuccess(response);
+    }
+
+    @Override
+    public void onFailure(Response response, Throwable failure) {
+        listener.onFailure(response, failure);
+    }
+}

apm-sniffer/apm-sdk-plugin/jetty-plugin/jetty-client-9.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/jetty/v9/client/AsyncHttpRequestSendInterceptorTest.java

@@ -33,6 +33,7 @@
 import org.apache.skywalking.apm.agent.test.tools.TracingSegmentRunner;
 import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
 import org.eclipse.jetty.client.HttpClient;
+import org.eclipse.jetty.client.HttpConversation;
 import org.eclipse.jetty.client.HttpRequest;
 import org.eclipse.jetty.client.ResponseNotifier;
 import org.eclipse.jetty.client.api.Response;
@@ -82,8 +83,8 @@ public class AsyncHttpRequestSendInterceptorTest {
 
     @Before
     public void setUp() throws Exception {
-        httpRequestEnhancedInstance = new MockHttpRequest(httpClient, uri);
-        responseNotifierEnhancedInstance = new MockResponseNotifier(httpClient);
+        httpRequestEnhancedInstance = new MockHttpRequest(httpClient, new HttpConversation(), uri);
+        responseNotifierEnhancedInstance = new MockResponseNotifier();
 
         Result results = new Result(httpRequestEnhancedInstance, response);
         allArguments = new Object[]{(Response.CompleteListener) result -> { }, results};
@@ -146,8 +147,8 @@ private void assertJettySpan() {
     }
 
     private class MockHttpRequest extends HttpRequest implements EnhancedInstance {
-        public MockHttpRequest(HttpClient httpClient, URI uri) {
-            super(httpClient, uri);
+        public MockHttpRequest(HttpClient client, HttpConversation conversation, URI uri) {
+            super(httpClient, conversation, uri);
         }
 
         @Override
@@ -172,8 +173,8 @@ public void setSkyWalkingDynamicField(Object value) {
     }
 
     private class MockResponseNotifier extends ResponseNotifier implements EnhancedInstance {
-        public MockResponseNotifier(HttpClient client) {
-            super(client);
+        public MockResponseNotifier() {
+            super();
         }
 
         @Override

apm-sniffer/apm-sdk-plugin/jetty-plugin/jetty-client-9.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/jetty/v9/client/SyncHttpRequestSendInterceptorTest.java

@@ -34,6 +34,7 @@
 import org.apache.skywalking.apm.agent.test.tools.SpanAssert;
 import org.apache.skywalking.apm.agent.test.tools.TracingSegmentRunner;
 import org.eclipse.jetty.client.HttpClient;
+import org.eclipse.jetty.client.HttpConversation;
 import org.eclipse.jetty.client.HttpRequest;
 import org.junit.Assert;
 import org.junit.Before;
@@ -67,7 +68,7 @@ public class SyncHttpRequestSendInterceptorTest {
 
     @Before
     public void setUp() throws Exception {
-        enhancedInstance = new MockHttpRequest(httpClient, uri);
+        enhancedInstance = new MockHttpRequest(httpClient, new HttpConversation(), uri);
         allArguments = new Object[] {
             "OperationKey",
             "OperationValue"
@@ -123,8 +124,8 @@ public void testMethodsAroundError() throws Throwable {
     }
 
     private class MockHttpRequest extends HttpRequest implements EnhancedInstance {
-        public MockHttpRequest(HttpClient httpClient, URI uri) {
-            super(httpClient, uri);
+        public MockHttpRequest(HttpClient client, HttpConversation conversation, URI uri) {
+            super(httpClient, conversation, uri);
         }
 
         @Override