Fix http concurrency test with large responses (#5648)

* Debug http client concurrency test failures

* debugging

* context porpagation to callbacks isn't really implemented

* verify that request succeeds in single connection concurrency test

* spotless

* verify request status in http client concurrency test

* update comment

* remove large response

* Trigger Build

* Update instrumentation/netty/netty-3.8/javaagent/src/test/groovy/Netty38ClientTest.groovy

Co-authored-by: Trask Stalnaker <[email protected]>

Co-authored-by: Trask Stalnaker <[email protected]>

Author: laurit

instrumentation/executors/testing/src/main/java/io/opentelemetry/javaagent/instrumentation/javaconcurrent/AbstractExecutorServiceTest.java

@@ -161,8 +161,7 @@ protected final void executeAndCancelTasks(Function<U, Future<?>> task) {
         });
 
     // Just check there is a single trace, this test is primarily to make sure that scopes aren't
-    // leak on
-    // cancellation.
+    // leaked on cancellation.
     testing.waitAndAssertTraces(trace -> {});
   }
 }

instrumentation/netty/netty-3.8/javaagent/src/test/groovy/Netty38ClientTest.groovy

@@ -10,6 +10,8 @@ import com.ning.http.client.Request
 import com.ning.http.client.RequestBuilder
 import com.ning.http.client.Response
 import io.opentelemetry.api.common.AttributeKey
+import io.opentelemetry.context.Context
+import io.opentelemetry.context.Scope
 import io.opentelemetry.instrumentation.test.AgentTestTrait
 import io.opentelemetry.instrumentation.test.base.HttpClientTest
 import io.opentelemetry.instrumentation.testing.junit.http.AbstractHttpClientTest
@@ -70,17 +72,23 @@ class Netty38ClientTest extends HttpClientTest<Request> implements AgentTestTrai
 
   @Override
   void sendRequestWithCallback(Request request, String method, URI uri, Map<String, String> headers, AbstractHttpClientTest.RequestResult requestResult) {
+    // TODO: context is not automatically propagated into callbacks
+    Context context = Context.current()
     // TODO(anuraaga): Do we also need to test ListenableFuture callback?
     client.executeRequest(request, new AsyncCompletionHandler<Void>() {
       @Override
       Void onCompleted(Response response) throws Exception {
-        requestResult.complete(response.statusCode)
+        try (Scope scope = context.makeCurrent()) {
+          requestResult.complete(response.statusCode)
+        }
         return null
       }
 
       @Override
       void onThrowable(Throwable throwable) {
-        requestResult.complete(throwable)
+        try (Scope scope = context.makeCurrent()) {
+          requestResult.complete(throwable)
+        }
       }
     })
   }

instrumentation/spring/spring-webflux-5.0/javaagent/src/test/groovy/client/SpringWebFluxSingleConnection.groovy

@@ -21,14 +21,15 @@ import java.util.concurrent.TimeoutException
 
 class SpringWebFluxSingleConnection implements SingleConnection {
   private final ReactorClientHttpConnector connector
+  private final WebClient webClient
   private final String host
   private final int port
 
   SpringWebFluxSingleConnection(boolean isOldVersion, String host, int port) {
     if (isOldVersion) {
       connector = new ReactorClientHttpConnector({ HttpClientOptions.Builder clientOptions ->
         clientOptions.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, HttpClientTest.CONNECT_TIMEOUT_MS)
-        clientOptions.poolResources(PoolResources.fixed("pool", 1, HttpClientTest.CONNECT_TIMEOUT_MS))
+        clientOptions.poolResources(PoolResources.fixed("pool", 1))
       })
     } else {
       def httpClient = HttpClient.create().tcpConfiguration({ tcpClient ->
@@ -40,6 +41,7 @@ class SpringWebFluxSingleConnection implements SingleConnection {
 
     this.host = host
     this.port = port
+    this.webClient = WebClient.builder().clientConnector(connector).build()
   }
 
   @Override
@@ -53,11 +55,13 @@ class SpringWebFluxSingleConnection implements SingleConnection {
       throw new ExecutionException(e)
     }
 
-    def request = WebClient.builder().clientConnector(connector).build().method(HttpMethod.GET)
+    def request = webClient.method(HttpMethod.GET)
       .uri(uri)
       .headers { h -> headers.forEach({ key, value -> h.add(key, value) }) }
 
     def response = request.exchange().block()
+    // read response body, this seems to be needed to ensure that the connection can be reused
+    response.bodyToMono(String).block()
 
     String responseId = response.headers().asHttpHeaders().getFirst(REQUEST_ID_HEADER)
     if (requestId != responseId) {

testing-common/src/main/java/io/opentelemetry/instrumentation/testing/junit/http/AbstractHttpClientTest.java

@@ -680,17 +680,22 @@ void highConcurrency() {
               throw new AssertionError(e);
             }
             try {
-              testing.runWithSpan(
-                  "Parent span " + index,
-                  () -> {
-                    Span.current().setAttribute("test.request.id", index);
-                    doRequest(
-                        method,
-                        uri,
-                        Collections.singletonMap("test-request-id", String.valueOf(index)));
-                  });
-            } catch (Exception e) {
-              throw new AssertionError(e);
+              Integer result =
+                  testing.runWithSpan(
+                      "Parent span " + index,
+                      () -> {
+                        Span.current().setAttribute("test.request.id", index);
+                        return doRequest(
+                            method,
+                            uri,
+                            Collections.singletonMap("test-request-id", String.valueOf(index)));
+                      });
+              assertThat(result).isEqualTo(200);
+            } catch (Throwable throwable) {
+              if (throwable instanceof AssertionError) {
+                throw (AssertionError) throwable;
+              }
+              throw new AssertionError(throwable);
             }
           };
       pool.submit(job);
@@ -832,19 +837,23 @@ void highConcurrencyOnSingleConnection() {
             } catch (InterruptedException e) {
               throw new AssertionError(e);
             }
-            testing.runWithSpan(
-                "Parent span " + index,
-                () -> {
-                  Span.current().setAttribute("test.request.id", index);
-                  try {
-                    singleConnection.doRequest(
-                        path, Collections.singletonMap("test-request-id", String.valueOf(index)));
-                  } catch (InterruptedException e) {
-                    throw new AssertionError(e);
-                  } catch (Exception e) {
-                    throw new AssertionError(e);
-                  }
-                });
+            try {
+              Integer result =
+                  testing.runWithSpan(
+                      "Parent span " + index,
+                      () -> {
+                        Span.current().setAttribute("test.request.id", index);
+                        return singleConnection.doRequest(
+                            path,
+                            Collections.singletonMap("test-request-id", String.valueOf(index)));
+                      });
+              assertThat(result).isEqualTo(200);
+            } catch (Throwable throwable) {
+              if (throwable instanceof AssertionError) {
+                throw (AssertionError) throwable;
+              }
+              throw new AssertionError(throwable);
+            }
           };
       pool.submit(job);
     }