Streamline Data Provider execution

Closes #3081

CHANGES.txt

@@ -1,4 +1,6 @@
 Current (7.10.0)
+
+Fixed: GITHUB-3081: Discrepancy with combination of (Shared Thread pool and Method Interceptor) (Krishnan Mahadevan)
 Fixed: GITHUB-2381: Controlling the inclusion of the listener at runtime (Krishnan Mahadevan)
 Fixed: GITHUB-3082: IInvokedMethodListener Iinvoked method does not return correct instance during @BeforeMethod, @AfterMethod and @AfterClass (Krishnan Mahadevan)
 Fixed: GITHUB-3084: Document project's PGP artifact signing keys (Krishnan Mahadevan)

testng-core/src/main/java/org/testng/internal/invokers/MethodRunner.java

@@ -1,7 +1,6 @@
 package org.testng.internal.invokers;
 
 import static java.util.concurrent.CompletableFuture.allOf;
-import static java.util.concurrent.CompletableFuture.supplyAsync;
 
 import java.util.ArrayList;
 import java.util.Collection;
@@ -23,6 +22,7 @@
 import org.testng.internal.Parameters;
 import org.testng.internal.invokers.ITestInvoker.FailureContext;
 import org.testng.internal.invokers.TestMethodArguments.Builder;
+import org.testng.internal.thread.Async;
 import org.testng.internal.thread.TestNGThreadFactory;
 import org.testng.xml.XmlSuite;
 
@@ -140,7 +140,7 @@ public List<ITestResult> runInParallel(
               invocationCount.get(),
               failure.count.get(),
               testInvoker.getNotifier());
-      all.add(supplyAsync(w::call, service));
+      all.add(Async.run(w, service));
       // testng387: increment the param index in the bag.
       parametersIndex += 1;
     }

testng-core/src/main/java/org/testng/internal/invokers/TestMethodWithDataProviderMethodWorker.java

@@ -14,7 +14,8 @@
 import org.testng.internal.invokers.TestMethodArguments.Builder;
 import org.testng.xml.XmlSuite;
 
-public class TestMethodWithDataProviderMethodWorker implements Callable<List<ITestResult>> {
+public class TestMethodWithDataProviderMethodWorker
+    implements Callable<List<ITestResult>>, Comparable<TestMethodWithDataProviderMethodWorker> {
 
   private final ITestNGMethod m_testMethod;
   private final Object[] m_parameterValues;
@@ -152,4 +153,9 @@ public List<ITestResult> call() {
   public int getInvocationCount() {
     return m_invocationCount;
   }
+
+  @Override
+  public int compareTo(TestMethodWithDataProviderMethodWorker o) {
+    return Integer.compare(this.m_testMethod.getPriority(), o.m_testMethod.getPriority());
+  }
 }

testng-core/src/main/java/org/testng/internal/thread/Async.java

@@ -0,0 +1,46 @@
+package org.testng.internal.thread;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import org.testng.ITestResult;
+import org.testng.internal.invokers.TestMethodWithDataProviderMethodWorker;
+
+public final class Async {
+
+  private Async() {
+    // Defeat instantiation
+  }
+
+  public static CompletableFuture<List<ITestResult>> run(
+      TestMethodWithDataProviderMethodWorker worker, ExecutorService service) {
+    AsyncTask asyncTask = new AsyncTask(worker);
+    service.execute(asyncTask);
+    return asyncTask.result;
+  }
+
+  private static class AsyncTask implements Runnable, Comparable<AsyncTask> {
+    private final CompletableFuture<List<ITestResult>> result = new CompletableFuture<>();
+    private final TestMethodWithDataProviderMethodWorker worker;
+
+    public AsyncTask(TestMethodWithDataProviderMethodWorker worker) {
+      this.worker = worker;
+    }
+
+    @Override
+    public void run() {
+      try {
+        if (!result.isDone()) {
+          result.complete(worker.call());
+        }
+      } catch (Throwable t) {
+        result.completeExceptionally(t);
+      }
+    }
+
+    @Override
+    public int compareTo(AsyncTask o) {
+      return worker.compareTo(o.worker);
+    }
+  }
+}

testng-core/src/test/java/test/dataprovider/DataProviderTest.java

@@ -59,6 +59,8 @@
 import test.dataprovider.issue3045.DataProviderListener;
 import test.dataprovider.issue3045.DataProviderTestClassSample;
 import test.dataprovider.issue3045.DataProviderWithoutListenerTestClassSample;
+import test.dataprovider.issue3081.NoOpMethodInterceptor;
+import test.dataprovider.issue3081.TestClassWithPrioritiesSample;
 
 public class DataProviderTest extends SimpleBaseTest {
 
@@ -651,6 +653,45 @@ public void ensureWeCanShareThreadPoolForDataProvidersThroughSuiteFiles(
         .hasSize(pair.second());
   }
 
+  @Test(description = "GITHUB-3081")
+  public void ensureNoExceptionsWhenRunningInSharedThreadPoolsWithMethodInterceptorsNoPriorities() {
+    TestNG testng = create(test.dataprovider.issue3081.TestClassSample.class);
+    test.dataprovider.issue3081.TestClassSample.clear();
+    testng.shouldUseGlobalThreadPool(true);
+    testng.addListener(new NoOpMethodInterceptor());
+    testng.setThreadCount(10);
+    testng.setParallel(XmlSuite.ParallelMode.METHODS);
+    testng.shareThreadPoolForDataProviders(true);
+    testng.setVerbose(2);
+    testng.run();
+    assertThat(testng.getStatus()).isEqualTo(0);
+    assertThat(test.dataprovider.issue3081.TestClassSample.getLogs())
+        .withFailMessage(
+            "There should have been 9 threads ONLY used by the data driven test "
+                + "because one thread would be the main thread on which TestNG would be running")
+        .hasSize(9);
+  }
+
+  @Test(description = "GITHUB-3081")
+  public void
+      ensureNoExceptionsWhenRunningInSharedThreadPoolsWithMethodInterceptorsWithPriorities() {
+    TestNG testng = create(TestClassWithPrioritiesSample.class);
+    TestClassWithPrioritiesSample.clear();
+    testng.shouldUseGlobalThreadPool(true);
+    testng.addListener(new NoOpMethodInterceptor());
+    testng.setParallel(XmlSuite.ParallelMode.METHODS);
+    testng.shareThreadPoolForDataProviders(true);
+    testng.setThreadCount(10);
+    testng.setVerbose(2);
+    testng.run();
+    assertThat(testng.getStatus()).isEqualTo(0);
+    assertThat(TestClassWithPrioritiesSample.getLogs())
+        .withFailMessage(
+            "There should have been 9 threads ONLY used by the data driven test "
+                + "because one thread would be the main thread on which TestNG would be running")
+        .hasSize(9);
+  }
+
   @DataProvider
   public Object[][] getSuiteFileNames() {
     return new Object[][] {

testng-core/src/test/java/test/dataprovider/issue3081/NoOpMethodInterceptor.java

@@ -0,0 +1,13 @@
+package test.dataprovider.issue3081;
+
+import java.util.List;
+import org.testng.IMethodInstance;
+import org.testng.IMethodInterceptor;
+import org.testng.ITestContext;
+
+public class NoOpMethodInterceptor implements IMethodInterceptor {
+  @Override
+  public List<IMethodInstance> intercept(List<IMethodInstance> methods, ITestContext context) {
+    return methods;
+  }
+}

testng-core/src/test/java/test/dataprovider/issue3081/TestClassSample.java

@@ -0,0 +1,39 @@
+package test.dataprovider.issue3081;
+
+import java.security.SecureRandom;
+import java.util.Collections;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+public class TestClassSample {
+  private static final Set<Long> logs = ConcurrentHashMap.newKeySet();
+  private static final Random random = new SecureRandom();
+
+  public static Set<Long> getLogs() {
+    return Collections.unmodifiableSet(logs);
+  }
+
+  public static void clear() {
+    logs.clear();
+  }
+
+  @DataProvider(parallel = true)
+  public static Object[] parallelDpStrings() {
+    return IntStream.rangeClosed(0, 99).mapToObj(it -> "string " + it).toArray(String[]::new);
+  }
+
+  @Test(dataProvider = "parallelDpStrings")
+  public void testStrings(String ignored) throws InterruptedException {
+    print();
+    TimeUnit.MILLISECONDS.sleep(random.nextInt(500));
+  }
+
+  private static void print() {
+    logs.add(Thread.currentThread().getId());
+  }
+}

testng-core/src/test/java/test/dataprovider/issue3081/TestClassWithPrioritiesSample.java

@@ -0,0 +1,42 @@
+package test.dataprovider.issue3081;
+
+import java.security.SecureRandom;
+import java.util.Collections;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+public class TestClassWithPrioritiesSample {
+  private static final Set<Long> logs = ConcurrentHashMap.newKeySet();
+  private static final Random random = new SecureRandom();
+
+  public static Set<Long> getLogs() {
+    return Collections.unmodifiableSet(logs);
+  }
+
+  public static void clear() {
+    logs.clear();
+  }
+
+  @DataProvider(parallel = true)
+  public static Object[] parallelDpStrings() {
+    return IntStream.rangeClosed(0, 99).mapToObj(it -> "string " + it).toArray(String[]::new);
+  }
+
+  @Test(dataProvider = "parallelDpStrings", priority = 1)
+  public void testStrings(String ignored) throws InterruptedException {
+    print();
+    TimeUnit.MILLISECONDS.sleep(random.nextInt(500));
+  }
+
+  @Test(priority = 2)
+  public void anotherTest() {}
+
+  private static void print() {
+    logs.add(Thread.currentThread().getId());
+  }
+}