Test http server requests in parallel (#8222)

* Test http server requests in parallel

* use executor service

* exclude some parallel

Author: amarziali

dd-java-agent/instrumentation/servlet/request-3/src/test/groovy/JettyServlet3Test.groovy

@@ -438,6 +438,11 @@ class JettyServlet3TestSyncDispatchOnAsyncTimeout extends JettyServlet3Test {
     true
   }
 
+  @Override
+  boolean testParallelRequest() {
+    false
+  }
+
   @Override
   void handlerSpan(TraceAssert trace, ServerEndpoint endpoint = SUCCESS) {
     dispatchSpan(trace, endpoint)
@@ -474,6 +479,11 @@ class JettyServlet3TestAsyncDispatchOnAsyncTimeout extends JettyServlet3Test {
     true
   }
 
+  @Override
+  boolean testParallelRequest() {
+    false
+  }
+
   @Override
   void handlerSpan(TraceAssert trace, ServerEndpoint endpoint = SUCCESS) {
     dispatchSpan(trace, endpoint)
@@ -532,6 +542,11 @@ class JettyServlet3ServeFromAsyncTimeout extends JettyServlet3Test {
   boolean testException() {
     false
   }
+
+  @Override
+  boolean testParallelRequest() {
+    false
+  }
 }
 
 class IastJettyServlet3ForkedTest extends JettyServlet3TestSync {

dd-java-agent/testing/src/main/groovy/datadog/trace/agent/test/base/HttpServerTest.groovy

@@ -47,6 +47,8 @@ import org.slf4j.Logger
 import org.slf4j.LoggerFactory
 
 import javax.annotation.Nonnull
+import java.util.concurrent.ExecutorCompletionService
+import java.util.concurrent.Executors
 import java.util.function.BiFunction
 import java.util.function.Function
 import java.util.function.Supplier
@@ -382,6 +384,10 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
     false // not all servers support session ids
   }
 
+  boolean testParallelRequest() {
+    true
+  }
+
   @Override
   int version() {
     return 0
@@ -528,10 +534,21 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
   @Flaky(value = "https://github.com/DataDog/dd-trace-java/issues/4690", suites = ["MuleHttpServerForkedTest"])
   def "test success with #count requests"() {
     setup:
+    def responses
     def request = request(SUCCESS, method, body).build()
-    List<Response> responses = (1..count).collect {
-      return client.newCall(request).execute()
+    if (testParallelRequest()) {
+      def executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors())
+      def completionService = new ExecutorCompletionService(executor)
+      (1..count).each {
+        completionService.submit {
+          client.newCall(request).execute()
+        }
+      }
+      responses = (1..count).collect { completionService.take().get() }
+    } else {
+      responses = (1..count).collect {client.newCall(request).execute()}
     }
+
     if (isDataStreamsEnabled()) {
       TEST_DATA_STREAMS_WRITER.waitForGroups(1)
     }