jetty-httpclient-12: send method must pass along implemented response listeners

The super HttpRequest.send call says:

> The listener passed to this method may implement not only Response.CompleteListener
> but also other response listener interfaces, and all the events implemented will be notified.

The current implementation passes through a lambda that implements CompleteListener only, and
does not delegate-in-scope to any other callback methods you might provide

Author: stevenschlansker

instrumentation/jetty-httpclient/jetty-httpclient-12.0/library/src/main/java/io/opentelemetry/instrumentation/jetty/httpclient/v12_0/TracingHttpRequest.java

@@ -9,6 +9,7 @@
 import io.opentelemetry.context.Scope;
 import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
 import io.opentelemetry.instrumentation.jetty.httpclient.v12_0.internal.JettyClientTracingListener;
+import io.opentelemetry.instrumentation.jetty.httpclient.v12_0.internal.JettyClientWrapUtil;
 import java.net.URI;
 import java.nio.ByteBuffer;
 import org.eclipse.jetty.client.HttpClient;
@@ -34,14 +35,9 @@ public TracingHttpRequest(
   @Override
   public void send(Response.CompleteListener listener) {
     parentContext = Context.current();
-    // start span and attach listeners.
+    // start span and attach listeners - must handle all listeners, not just CompleteListener.
     JettyClientTracingListener.handleRequest(parentContext, this, instrumenter);
-    super.send(
-        result -> {
-          try (Scope scope = openScope()) {
-            listener.onComplete(result);
-          }
-        });
+    super.send(JettyClientWrapUtil.wrapTheListener(listener, parentContext));
   }
 
   private Scope openScope() {

instrumentation/jetty-httpclient/jetty-httpclient-12.0/library/src/main/java/io/opentelemetry/instrumentation/jetty/httpclient/v12_0/internal/JettyClientWrapUtil.java

@@ -0,0 +1,71 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.jetty.httpclient.v12_0.internal;
+
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.Scope;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Proxy;
+import java.util.ArrayList;
+import java.util.List;
+import org.eclipse.jetty.client.Response;
+
+/**
+ * This class is internal and is hence not for public use. Its APIs are unstable and can change at
+ * any time.
+ */
+public final class JettyClientWrapUtil {
+  private static final Class<?>[] LISTENER_INTERFACES = {
+    Response.CompleteListener.class,
+    Response.FailureListener.class,
+    Response.SuccessListener.class,
+    Response.AsyncContentListener.class,
+    Response.ContentSourceListener.class,
+    Response.ContentListener.class,
+    Response.HeadersListener.class,
+    Response.HeaderListener.class,
+    Response.BeginListener.class
+  };
+
+  private JettyClientWrapUtil() {}
+
+  /**
+   * Utility to wrap the response listeners only, this includes the important CompleteListener.
+   *
+   * @param context top level context that is above the Jetty client span context
+   * @param listener listener passed to Jetty client send() method
+   * @return wrapped listener
+   */
+  public static Response.CompleteListener wrapTheListener(
+      Response.CompleteListener listener, Context context) {
+    if (listener == null) {
+      return listener;
+    }
+
+    Class<?> listenerClass = listener.getClass();
+    List<Class<?>> interfaces = new ArrayList<>();
+    for (Class<?> type : LISTENER_INTERFACES) {
+      if (type.isInstance(listener)) {
+        interfaces.add(type);
+      }
+    }
+    if (interfaces.isEmpty()) {
+      return listener;
+    }
+
+    return (Response.CompleteListener)
+        Proxy.newProxyInstance(
+            listenerClass.getClassLoader(),
+            interfaces.toArray(new Class<?>[0]),
+            (proxy, method, args) -> {
+              try (Scope ignored = context.makeCurrent()) {
+                return method.invoke(listener, args);
+              } catch (InvocationTargetException exception) {
+                throw exception.getCause();
+              }
+            });
+  }
+}

instrumentation/jetty-httpclient/jetty-httpclient-12.0/testing/src/main/java/io/opentelemetry/instrumentation/jetty/httpclient/v12_0/AbstractJettyClient12Test.java

@@ -5,23 +5,30 @@
 
 package io.opentelemetry.instrumentation.jetty.httpclient.v12_0;
 
+import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.assertThat;
+
 import io.opentelemetry.instrumentation.testing.junit.http.AbstractHttpClientTest;
 import io.opentelemetry.instrumentation.testing.junit.http.HttpClientResult;
 import io.opentelemetry.instrumentation.testing.junit.http.HttpClientTestOptions;
 import java.net.URI;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import org.eclipse.jetty.client.ContentResponse;
+import org.eclipse.jetty.client.FutureResponseListener;
 import org.eclipse.jetty.client.HttpClient;
 import org.eclipse.jetty.client.Request;
 import org.eclipse.jetty.client.Response;
+import org.eclipse.jetty.client.Result;
 import org.eclipse.jetty.http.HttpField;
 import org.eclipse.jetty.util.ssl.SslContextFactory;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
 
 public abstract class AbstractJettyClient12Test extends AbstractHttpClientTest<Request> {
 
@@ -108,6 +115,38 @@ public void sendRequestWithCallback(
         });
   }
 
+  @Test
+  void callbacksCalled() throws InterruptedException, ExecutionException {
+    URI uri = resolveAddress("/success");
+    Set<String> callbacks = ConcurrentHashMap.newKeySet();
+    Request request = client.newRequest(uri).method("GET");
+    FutureResponseListener responseListener =
+        new FutureResponseListener(request) {
+          @Override
+          public void onHeaders(Response response) {
+            callbacks.add("headers");
+            super.onHeaders(response);
+          }
+
+          @Override
+          public void onSuccess(Response response) {
+            callbacks.add("success");
+            super.onSuccess(response);
+          }
+
+          @Override
+          public void onComplete(Result result) {
+            callbacks.add("complete");
+            super.onComplete(result);
+          }
+        };
+    request.send(responseListener);
+    Response response = responseListener.get();
+
+    assertThat(response.getStatus()).isEqualTo(200);
+    assertThat(callbacks).containsExactlyInAnyOrder("headers", "success", "complete");
+  }
+
   private static class JettyClientListener
       implements Request.FailureListener, Response.FailureListener {
     volatile Throwable failure;