diff --git a/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Dataflow.json b/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Dataflow.json index 8144784f5f02..97b7503f5445 100644 --- a/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Dataflow.json +++ b/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Dataflow.json @@ -1,5 +1,5 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run!", - "modification": 2, + "modification": 7, } diff --git a/runners/google-cloud-dataflow-java/build.gradle b/runners/google-cloud-dataflow-java/build.gradle index af42dc2a5013..92e6963c3eab 100644 --- a/runners/google-cloud-dataflow-java/build.gradle +++ b/runners/google-cloud-dataflow-java/build.gradle @@ -236,16 +236,31 @@ def commonRunnerV2ExcludeCategories = [ 'org.apache.beam.sdk.testing.UsesBoundedTrieMetrics', // Dataflow QM as of now does not support returning back BoundedTrie in metric result. ] + +def validatesRunnerParallelThreads = project.findProperty('parallelThreads') ?: '6' + def isValidatesRunnerTestClass = { FileTreeElement element -> if (element.isDirectory()) { return true } if (!element.name.endsWith('.class') || - element.name.startsWith('ValidateRunnerXlangTest')) { + element.name.startsWith('ValidateRunnerXlangTest') || + element.name.startsWith('BeamParallelJunit4RunnerTest')) { return false } - return new String(element.file.bytes, java.nio.charset.StandardCharsets.ISO_8859_1) - .contains('Lorg/apache/beam/sdk/testing/ValidatesRunner;') + def content = new String(element.file.bytes, java.nio.charset.StandardCharsets.ISO_8859_1) + if (content.contains('Lorg/apache/beam/sdk/testing/ValidatesRunner;')) { + return true + } + if (content.contains('Lorg/junit/experimental/runners/Enclosed;')) { + def prefix = element.name.substring(0, element.name.length() - 6) + '$' + return element.file.parentFile.listFiles()?.any { File f -> + f.name.startsWith(prefix) && f.name.endsWith('.class') && + new String(f.bytes, java.nio.charset.StandardCharsets.ISO_8859_1) + .contains('Lorg/apache/beam/sdk/testing/ValidatesRunner;') + } ?: false + } + return false } def createLegacyWorkerValidatesRunnerTest = { Map args -> @@ -258,10 +273,12 @@ def createLegacyWorkerValidatesRunnerTest = { Map args -> dependsOn ":runners:google-cloud-dataflow-java:worker:shadowJar" systemProperty "beamTestPipelineOptions", JsonOutput.toJson(pipelineOptions) + systemProperty "beam.validatesRunner.parallelThreads", validatesRunnerParallelThreads // Increase test parallelism up to the number of Gradle workers. By default this is equal // to the number of CPU cores, but can be increased by setting --max-workers=N. maxParallelForks Integer.MAX_VALUE + maxHeapSize = '2g' classpath = configurations.validatesRunner testClassesDirs = files(project(":sdks:java:core").sourceSets.test.output.classesDirs) + files(project(project.path).sourceSets.test.output.classesDirs) @@ -292,10 +309,12 @@ def createRunnerV2ValidatesRunnerTest = { Map args -> return tasks.create(name: name, type: Test, group: "Verification") { dependsOn buildAndPushDockerJavaContainer systemProperty "beamTestPipelineOptions", JsonOutput.toJson(pipelineOptions) + systemProperty "beam.validatesRunner.parallelThreads", validatesRunnerParallelThreads // Increase test parallelism up to the number of Gradle workers. By default this is equal // to the number of CPU cores, but can be increased by setting --max-workers=N. maxParallelForks Integer.MAX_VALUE + maxHeapSize = '768m' classpath = configurations.validatesRunner testClassesDirs = files(project(":sdks:java:core").sourceSets.test.output.classesDirs) + files(project(project.path).sourceSets.test.output.classesDirs) diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/BeamParallelJunit4Runner.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/BeamParallelJunit4Runner.java new file mode 100644 index 000000000000..bcaead005add --- /dev/null +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/BeamParallelJunit4Runner.java @@ -0,0 +1,189 @@ +/* + * 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.beam.sdk.testing; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.beam.sdk.annotations.Internal; +import org.checkerframework.checker.nullness.qual.Nullable; +import org.junit.experimental.categories.Category; +import org.junit.runner.notification.RunNotifier; +import org.junit.runners.BlockJUnit4ClassRunner; +import org.junit.runners.model.FrameworkMethod; +import org.junit.runners.model.InitializationError; +import org.junit.runners.model.RunnerScheduler; + +/** + * A JUnit 4 runner that supports concurrent execution of {@code @Test} methods within a test class, + * with independent parallelism settings for {@link ValidatesRunner} tests versus other tests: + * + * + * + *

Classes or methods annotated with {@link SerialTest} are always executed sequentially on the + * calling thread after draining any in-flight parallel test methods in the class. + */ +@Internal +public final class BeamParallelJunit4Runner extends BlockJUnit4ClassRunner { + + /** + * Marks a test class or {@code @Test} method as requiring serial (non-parallel) execution even + * when parallel test execution is enabled. + */ + @Retention(RetentionPolicy.RUNTIME) + @Target({ElementType.TYPE, ElementType.METHOD}) + public @interface SerialTest {} + + public static final String VALIDATES_RUNNER_THREADS_PROPERTY = + "beam.validatesRunner.parallelThreads"; + public static final String DEFAULT_TEST_THREADS_PROPERTY = "beam.test.parallelThreads"; + + private static final ConcurrentHashMap EXECUTORS = + new ConcurrentHashMap<>(); + + private static final ConcurrentHashMap, Boolean> CLASS_SERIAL_CACHE = + new ConcurrentHashMap<>(); + + private CompletableFuture pendingFutures = CompletableFuture.allOf(); + + private static @Nullable ExecutorService getOrCreateExecutor(String poolKey, int threads) { + if (threads <= 1) { + return null; + } + return EXECUTORS.computeIfAbsent( + poolKey + ":" + threads, + k -> { + AtomicInteger counter = new AtomicInteger(1); + return Executors.newFixedThreadPool( + threads, + runnable -> { + Thread thread = new Thread(runnable); + thread.setDaemon(true); + thread.setName(poolKey + "-" + counter.getAndIncrement()); + return thread; + }); + }); + } + + public BeamParallelJunit4Runner(Class klass) throws InitializationError { + super(klass); + setScheduler( + new RunnerScheduler() { + @Override + public void schedule(Runnable childStatement) { + childStatement.run(); + } + + @Override + public void finished() { + awaitPendingFutures(); + } + }); + } + + private void awaitPendingFutures() { + try { + pendingFutures.join(); + } finally { + pendingFutures = CompletableFuture.allOf(); + } + } + + @Override + protected void runChild(final FrameworkMethod method, final RunNotifier notifier) { + if (isIgnored(method)) { + super.runChild(method, notifier); + return; + } + ExecutorService executor = selectExecutor(method); + if (executor == null) { + super.runChild(method, notifier); + return; + } + if (isClassMarkedSerial(getTestClass().getJavaClass()) || isMethodMarkedSerial(method)) { + // This is a serial test. Make sure to wait for any in-flight tests to complete, and then run + // this test + // serially. + awaitPendingFutures(); + super.runChild(method, notifier); + return; + } + pendingFutures = + CompletableFuture.allOf( + pendingFutures, + CompletableFuture.runAsync(() -> super.runChild(method, notifier), executor)); + } + + private @Nullable ExecutorService selectExecutor(FrameworkMethod method) { + int defaultThreads = Integer.getInteger(DEFAULT_TEST_THREADS_PROPERTY, 1); + if (isValidatesRunnerMethod(method)) { + int vrThreads = Integer.getInteger(VALIDATES_RUNNER_THREADS_PROPERTY, defaultThreads); + return getOrCreateExecutor("beam-vr-worker", vrThreads); + } + return getOrCreateExecutor("beam-test-worker", defaultThreads); + } + + private boolean isValidatesRunnerMethod(FrameworkMethod method) { + return hasValidatesRunnerCategory(method.getAnnotation(Category.class)) + || hasValidatesRunnerCategory(getTestClass().getAnnotation(Category.class)); + } + + private static boolean hasValidatesRunnerCategory(@Nullable Category category) { + if (category == null) { + return false; + } + for (Class c : category.value()) { + if (ValidatesRunner.class.isAssignableFrom(c)) { + return true; + } + } + return false; + } + + private static boolean isMethodMarkedSerial(FrameworkMethod method) { + return method.getAnnotation(SerialTest.class) != null; + } + + private static boolean isClassMarkedSerial(@Nullable Class clazz) { + if (clazz == null || clazz == Object.class) { + return false; + } + return CLASS_SERIAL_CACHE.computeIfAbsent(clazz, BeamParallelJunit4Runner::computeClassSerial); + } + + private static boolean computeClassSerial(@Nullable Class clazz) { + if (clazz == null || clazz == Object.class) { + return false; + } + return clazz.isAnnotationPresent(SerialTest.class) + || computeClassSerial(clazz.getSuperclass()) + || computeClassSerial(clazz.getEnclosingClass()); + } +} diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/resources/ClasspathScanningResourcesDetector.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/resources/ClasspathScanningResourcesDetector.java index 5a5c6b325272..c1d815de6ab2 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/resources/ClasspathScanningResourcesDetector.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/resources/ClasspathScanningResourcesDetector.java @@ -82,7 +82,6 @@ public List detect(@Nullable ClassLoader classLoader) { if (snapshot != null && snapshot.matches(classLoader, currentJavaClassPath)) { return new ArrayList<>(snapshot.files); } - List classpathContents; if (classLoader != null) { classpathContents = diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/PipelineTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/PipelineTest.java index 4a4ef1dca6a1..a5c74b43ab34 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/PipelineTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/PipelineTest.java @@ -42,6 +42,7 @@ import org.apache.beam.sdk.runners.PTransformOverride; import org.apache.beam.sdk.runners.PTransformOverrideFactory; import org.apache.beam.sdk.runners.TransformHierarchy.Node; +import org.apache.beam.sdk.testing.BeamParallelJunit4Runner; import org.apache.beam.sdk.testing.CrashingRunner; import org.apache.beam.sdk.testing.ExpectedLogs; import org.apache.beam.sdk.testing.NeedsRunner; @@ -77,10 +78,9 @@ import org.junit.rules.ExpectedException; import org.junit.rules.Timeout; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; /** Tests for Pipeline. */ -@RunWith(JUnit4.class) +@RunWith(BeamParallelJunit4Runner.class) @SuppressWarnings({ "rawtypes", // TODO(https://github.com/apache/beam/issues/20447) }) @@ -273,6 +273,7 @@ public void testToString() { } @Test + @BeamParallelJunit4Runner.SerialTest public void testStableUniqueNameOff() { pipeline.enableAbandonedNodeEnforcement(false); @@ -285,6 +286,7 @@ public void testStableUniqueNameOff() { } @Test + @BeamParallelJunit4Runner.SerialTest public void testStableUniqueNameWarning() { pipeline.enableAbandonedNodeEnforcement(false); @@ -297,6 +299,7 @@ public void testStableUniqueNameWarning() { } @Test + @BeamParallelJunit4Runner.SerialTest public void testStableUniqueNameError() { pipeline.getOptions().setStableUniqueNames(CheckEnabled.ERROR); diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/SdkHarnessEnvironmentTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/SdkHarnessEnvironmentTest.java index f258529bee2b..a04a123eae15 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/SdkHarnessEnvironmentTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/SdkHarnessEnvironmentTest.java @@ -33,6 +33,7 @@ import org.apache.beam.sdk.options.SdkHarnessOptions; import org.apache.beam.sdk.options.SdkHarnessOptions.LogLevel; import org.apache.beam.sdk.options.SdkHarnessOptions.SdkHarnessLogLevelOverrides; +import org.apache.beam.sdk.testing.BeamParallelJunit4Runner; import org.apache.beam.sdk.testing.ExpectedLogs; import org.apache.beam.sdk.testing.ExpectedLogs.LogSaver; import org.apache.beam.sdk.testing.PAssert; @@ -49,10 +50,9 @@ import org.junit.Test; import org.junit.experimental.categories.Category; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; /** Tests that validate the SDK harness is configured correctly for a runner. */ -@RunWith(JUnit4.class) +@RunWith(BeamParallelJunit4Runner.class) public class SdkHarnessEnvironmentTest { @Rule public final TestPipeline p = TestPipeline.create(); diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/CountingSourceTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/CountingSourceTest.java index 337462340df1..b8542945d509 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/CountingSourceTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/CountingSourceTest.java @@ -31,6 +31,7 @@ import org.apache.beam.sdk.io.CountingSource.UnboundedCountingSource; import org.apache.beam.sdk.io.UnboundedSource.UnboundedReader; import org.apache.beam.sdk.options.PipelineOptionsFactory; +import org.apache.beam.sdk.testing.BeamParallelJunit4Runner; import org.apache.beam.sdk.testing.NeedsRunner; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; @@ -53,10 +54,9 @@ import org.junit.Test; import org.junit.experimental.categories.Category; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; /** Tests of {@link CountingSource}. */ -@RunWith(JUnit4.class) +@RunWith(BeamParallelJunit4Runner.class) public class CountingSourceTest { public static void addCountingAsserts(PCollection input, long numElements) { diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/MetricsTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/MetricsTest.java index 2fc800adca56..7e4589fdfa68 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/MetricsTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/MetricsTest.java @@ -42,6 +42,7 @@ import org.apache.beam.sdk.io.Read; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.options.PipelineOptionsFactory; +import org.apache.beam.sdk.testing.BeamParallelJunit4Runner; import org.apache.beam.sdk.testing.NeedsRunner; import org.apache.beam.sdk.testing.TestPipeline; import org.apache.beam.sdk.testing.UsesAttemptedMetrics; @@ -68,7 +69,6 @@ import org.junit.experimental.categories.Category; import org.junit.rules.ExpectedException; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; import org.mockito.Mockito; /** Tests for {@link Metrics}. */ @@ -176,7 +176,7 @@ public void processElement(ProcessContext c) { } /** Tests validating basic metric scenarios. */ - @RunWith(JUnit4.class) + @RunWith(BeamParallelJunit4Runner.class) public static class BasicTests extends SharedTestBase { @Test public void testDistributionWithoutContainer() { @@ -259,6 +259,7 @@ public void testCounterToCell() { } @Test + @BeamParallelJunit4Runner.SerialTest public void testMetricsFlag() { Metrics.resetDefaultPipelineOptions(); assertFalse(Metrics.MetricsFlag.counterDisabled()); @@ -293,7 +294,7 @@ public void testMetricsFlag() { } /** Tests for committed metrics. */ - @RunWith(JUnit4.class) + @RunWith(BeamParallelJunit4Runner.class) public static class CommittedMetricTests extends SharedTestBase { @Category({ ValidatesRunner.class, @@ -430,7 +431,7 @@ public void testUnboundedSourceMetrics() { } /** Tests for attempted metrics. */ - @RunWith(JUnit4.class) + @RunWith(BeamParallelJunit4Runner.class) public static class AttemptedMetricTests extends SharedTestBase { @Category({ ValidatesRunner.class, diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/BeamParallelJunit4RunnerTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/BeamParallelJunit4RunnerTest.java new file mode 100644 index 000000000000..1fd2b168bd06 --- /dev/null +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/BeamParallelJunit4RunnerTest.java @@ -0,0 +1,203 @@ +/* + * 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.beam.sdk.testing; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assume.assumeTrue; + +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.beam.sdk.testing.BeamParallelJunit4Runner.SerialTest; +import org.junit.Rule; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.rules.TestRule; +import org.junit.runner.JUnitCore; +import org.junit.runner.Result; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +/** Tests for {@link BeamParallelJunit4Runner}. */ +@RunWith(JUnit4.class) +public class BeamParallelJunit4RunnerTest { + + private static final AtomicBoolean IN_TEST_HARNESS = new AtomicBoolean(false); + + @Rule public final TestRule restoreProperties = new RestoreSystemProperties(); + + @RunWith(BeamParallelJunit4Runner.class) + public static class SampleMixedCases { + static final CountDownLatch VR_LATCH = new CountDownLatch(2); + static final Set VR_THREADS = ConcurrentHashMap.newKeySet(); + static final Set UNIT_THREADS = ConcurrentHashMap.newKeySet(); + static final AtomicInteger UNIT_ACTIVE = new AtomicInteger(0); + static final AtomicInteger UNIT_MAX_ACTIVE = new AtomicInteger(0); + + @Test + @Category(ValidatesRunner.class) + public void testVr1() throws InterruptedException { + assumeTrue(IN_TEST_HARNESS.get()); + VR_THREADS.add(Thread.currentThread().getName()); + VR_LATCH.countDown(); + assertTrue(VR_LATCH.await(5, TimeUnit.SECONDS)); + } + + @Test + @Category(ValidatesRunner.class) + public void testVr2() throws InterruptedException { + assumeTrue(IN_TEST_HARNESS.get()); + VR_THREADS.add(Thread.currentThread().getName()); + VR_LATCH.countDown(); + assertTrue(VR_LATCH.await(5, TimeUnit.SECONDS)); + } + + @Test + public void testUnit1() throws InterruptedException { + assumeTrue(IN_TEST_HARNESS.get()); + int active = UNIT_ACTIVE.incrementAndGet(); + UNIT_MAX_ACTIVE.accumulateAndGet(active, Math::max); + UNIT_THREADS.add(Thread.currentThread().getName()); + Thread.sleep(50); + UNIT_ACTIVE.decrementAndGet(); + } + + @Test + public void testUnit2() throws InterruptedException { + assumeTrue(IN_TEST_HARNESS.get()); + int active = UNIT_ACTIVE.incrementAndGet(); + UNIT_MAX_ACTIVE.accumulateAndGet(active, Math::max); + UNIT_THREADS.add(Thread.currentThread().getName()); + Thread.sleep(50); + UNIT_ACTIVE.decrementAndGet(); + } + } + + @SerialTest + @RunWith(BeamParallelJunit4Runner.class) + public static class SampleSerialClassCases { + static final AtomicInteger ACTIVE = new AtomicInteger(0); + static final AtomicInteger MAX_ACTIVE = new AtomicInteger(0); + + @Test + @Category(ValidatesRunner.class) + public void test1() throws InterruptedException { + assumeTrue(IN_TEST_HARNESS.get()); + int active = ACTIVE.incrementAndGet(); + MAX_ACTIVE.accumulateAndGet(active, Math::max); + Thread.sleep(50); + ACTIVE.decrementAndGet(); + } + + @Test + @Category(ValidatesRunner.class) + public void test2() throws InterruptedException { + assumeTrue(IN_TEST_HARNESS.get()); + int active = ACTIVE.incrementAndGet(); + MAX_ACTIVE.accumulateAndGet(active, Math::max); + Thread.sleep(50); + ACTIVE.decrementAndGet(); + } + } + + @RunWith(BeamParallelJunit4Runner.class) + public static class SampleSerialMethodCases { + static final AtomicInteger ACTIVE = new AtomicInteger(0); + static final AtomicInteger SERIAL_OBSERVED_ACTIVE = new AtomicInteger(0); + + @Test + @Category(ValidatesRunner.class) + public void test1Parallel() throws InterruptedException { + assumeTrue(IN_TEST_HARNESS.get()); + ACTIVE.incrementAndGet(); + Thread.sleep(60); + ACTIVE.decrementAndGet(); + } + + @Test + @SerialTest + @Category(ValidatesRunner.class) + public void test2Serial() throws InterruptedException { + assumeTrue(IN_TEST_HARNESS.get()); + int active = ACTIVE.incrementAndGet(); + SERIAL_OBSERVED_ACTIVE.accumulateAndGet(active, Math::max); + Thread.sleep(40); + ACTIVE.decrementAndGet(); + } + + @Test + @Category(ValidatesRunner.class) + public void test3Parallel() throws InterruptedException { + assumeTrue(IN_TEST_HARNESS.get()); + ACTIVE.incrementAndGet(); + Thread.sleep(60); + ACTIVE.decrementAndGet(); + } + } + + @Test + public void testSeparateParallelismForValidatesRunnerVsUnitTests() { + System.setProperty(BeamParallelJunit4Runner.VALIDATES_RUNNER_THREADS_PROPERTY, "2"); + System.setProperty(BeamParallelJunit4Runner.DEFAULT_TEST_THREADS_PROPERTY, "1"); + + IN_TEST_HARNESS.set(true); + try { + Result result = JUnitCore.runClasses(SampleMixedCases.class); + assertEquals(0, result.getFailureCount()); + assertEquals(4, result.getRunCount()); + assertEquals(2, SampleMixedCases.VR_THREADS.size()); + assertEquals(1, SampleMixedCases.UNIT_MAX_ACTIVE.get()); + } finally { + IN_TEST_HARNESS.set(false); + } + } + + @Test + public void testSerialTestClassRunsSerially() { + System.setProperty(BeamParallelJunit4Runner.VALIDATES_RUNNER_THREADS_PROPERTY, "4"); + + IN_TEST_HARNESS.set(true); + try { + Result result = JUnitCore.runClasses(SampleSerialClassCases.class); + assertEquals(0, result.getFailureCount()); + assertEquals(2, result.getRunCount()); + assertEquals(1, SampleSerialClassCases.MAX_ACTIVE.get()); + } finally { + IN_TEST_HARNESS.set(false); + } + } + + @Test + public void testSerialTestMethodDrainsAndRunsAlone() { + System.setProperty(BeamParallelJunit4Runner.VALIDATES_RUNNER_THREADS_PROPERTY, "4"); + + IN_TEST_HARNESS.set(true); + try { + Result result = JUnitCore.runClasses(SampleSerialMethodCases.class); + assertEquals(0, result.getFailureCount()); + assertEquals(3, result.getRunCount()); + assertEquals(1, SampleSerialMethodCases.SERIAL_OBSERVED_ACTIVE.get()); + } finally { + IN_TEST_HARNESS.set(false); + } + } +} diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/PAssertTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/PAssertTest.java index a02196bb2c05..bec3de5a6241 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/PAssertTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/PAssertTest.java @@ -70,10 +70,9 @@ import org.junit.experimental.categories.Category; import org.junit.rules.ExpectedException; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; /** Test case for {@link PAssert}. */ -@RunWith(JUnit4.class) +@RunWith(BeamParallelJunit4Runner.class) @SuppressWarnings({ "rawtypes", // TODO(https://github.com/apache/beam/issues/20447) }) diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/TestStreamTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/TestStreamTest.java index 00b6027f945d..e9fec99da073 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/TestStreamTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/TestStreamTest.java @@ -73,10 +73,9 @@ import org.junit.experimental.categories.Category; import org.junit.rules.ExpectedException; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; /** Tests for {@link TestStream}. */ -@RunWith(JUnit4.class) +@RunWith(BeamParallelJunit4Runner.class) // TODO(https://github.com/apache/beam/issues/21230): Remove when new version of errorprone is // released (2.11.0) @SuppressWarnings("unused") diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/CombineFnsTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/CombineFnsTest.java index 328bfba8f3ce..2575729b8853 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/CombineFnsTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/CombineFnsTest.java @@ -28,6 +28,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; import org.apache.beam.sdk.coders.AtomicCoder; import org.apache.beam.sdk.coders.BigEndianIntegerCoder; import org.apache.beam.sdk.coders.Coder; @@ -35,6 +36,7 @@ import org.apache.beam.sdk.coders.KvCoder; import org.apache.beam.sdk.coders.NullableCoder; import org.apache.beam.sdk.coders.StringUtf8Coder; +import org.apache.beam.sdk.testing.BeamParallelJunit4Runner; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; import org.apache.beam.sdk.testing.UsesSideInputs; @@ -56,10 +58,9 @@ import org.junit.experimental.categories.Category; import org.junit.rules.ExpectedException; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; /** Unit tests for {@link CombineFns}. */ -@RunWith(JUnit4.class) +@RunWith(BeamParallelJunit4Runner.class) public class CombineFnsTest { @Rule public final TestPipeline p = TestPipeline.create(); @Rule public ExpectedException expectedException = ExpectedException.none(); @@ -279,11 +280,11 @@ public String apply(String input) { private static class DisplayDataCombineFn extends Combine.CombineFn { private final String value; - private static int i; + private static final AtomicInteger NEXT_ID = new AtomicInteger(); private final int id; DisplayDataCombineFn(String value) { - id = ++i; + id = NEXT_ID.incrementAndGet(); this.value = value; } diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/CombineTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/CombineTest.java index 993b84a528d7..c888cadedf1f 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/CombineTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/CombineTest.java @@ -53,6 +53,7 @@ import org.apache.beam.sdk.coders.StringUtf8Coder; import org.apache.beam.sdk.coders.VarIntCoder; import org.apache.beam.sdk.coders.VoidCoder; +import org.apache.beam.sdk.testing.BeamParallelJunit4Runner; import org.apache.beam.sdk.testing.NeedsRunner; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; @@ -94,7 +95,6 @@ import org.junit.experimental.categories.Category; import org.junit.rules.Timeout; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; /** Tests for {@link Combine} transforms. */ public class CombineTest implements Serializable { @@ -649,7 +649,7 @@ private static PCollection> createInput( } /** Tests validating basic Combine transform scenarios. */ - @RunWith(JUnit4.class) + @RunWith(BeamParallelJunit4Runner.class) public static class BasicTests extends SharedTestBase { @Test @Category({ValidatesRunner.class, UsesSideInputs.class}) @@ -984,7 +984,7 @@ public void testLambdaDisplayData() { } /** Tests validating CombineWithContext behaviors. */ - @RunWith(JUnit4.class) + @RunWith(BeamParallelJunit4Runner.class) public static class CombineWithContextTests extends SharedTestBase { @Test @Category({ValidatesRunner.class, UsesSideInputs.class}) @@ -1056,7 +1056,7 @@ public void testHotKeyCombineWithSideInputs() { } /** Tests validating windowing behaviors. */ - @RunWith(JUnit4.class) + @RunWith(BeamParallelJunit4Runner.class) public static class WindowingTests extends SharedTestBase implements Serializable { @Test @Category({ValidatesRunner.class}) @@ -1460,7 +1460,7 @@ public void testCombineGloballyInstanceMethodReference() { } /** Tests validating accumulation scenarios. */ - @RunWith(JUnit4.class) + @RunWith(BeamParallelJunit4Runner.class) public static class AccumulationTests extends SharedTestBase { @Test @Category({ValidatesRunner.class, UsesSideInputs.class}) diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/CreateTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/CreateTest.java index 795c18f8ca12..2dffdf63f6b3 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/CreateTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/CreateTest.java @@ -51,6 +51,7 @@ import org.apache.beam.sdk.options.ValueProvider.StaticValueProvider; import org.apache.beam.sdk.schemas.Schema; import org.apache.beam.sdk.schemas.SchemaCoder; +import org.apache.beam.sdk.testing.BeamParallelJunit4Runner; import org.apache.beam.sdk.testing.NeedsRunner; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.SourceTestUtils; @@ -79,10 +80,9 @@ import org.junit.experimental.categories.Category; import org.junit.rules.ExpectedException; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; /** Tests for Create. */ -@RunWith(JUnit4.class) +@RunWith(BeamParallelJunit4Runner.class) @SuppressWarnings({ "unchecked", }) diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/FlattenTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/FlattenTest.java index 74fb2dc99c88..34154cb65eb0 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/FlattenTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/FlattenTest.java @@ -45,6 +45,7 @@ import org.apache.beam.sdk.coders.VarLongCoder; import org.apache.beam.sdk.coders.VoidCoder; import org.apache.beam.sdk.io.GenerateSequence; +import org.apache.beam.sdk.testing.BeamParallelJunit4Runner; import org.apache.beam.sdk.testing.FlattenWithHeterogeneousCoders; import org.apache.beam.sdk.testing.NeedsRunner; import org.apache.beam.sdk.testing.PAssert; @@ -72,10 +73,9 @@ import org.junit.rules.ExpectedException; import org.junit.rules.Timeout; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; /** Tests for Flatten. */ -@RunWith(JUnit4.class) +@RunWith(BeamParallelJunit4Runner.class) public class FlattenTest implements Serializable { @Rule public final transient TestPipeline p = TestPipeline.create(); diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyTest.java index 18541437d5fb..13af0c4b356f 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyTest.java @@ -55,6 +55,7 @@ import org.apache.beam.sdk.state.Timer; import org.apache.beam.sdk.state.TimerSpec; import org.apache.beam.sdk.state.TimerSpecs; +import org.apache.beam.sdk.testing.BeamParallelJunit4Runner; import org.apache.beam.sdk.testing.LargeKeys; import org.apache.beam.sdk.testing.NeedsRunner; import org.apache.beam.sdk.testing.PAssert; @@ -95,7 +96,6 @@ import org.junit.Test; import org.junit.experimental.categories.Category; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; /** Tests for GroupByKey. */ @SuppressWarnings({ @@ -110,7 +110,7 @@ public abstract static class SharedTestBase { } /** Tests validating basic {@link GroupByKey} scenarios. */ - @RunWith(JUnit4.class) + @RunWith(BeamParallelJunit4Runner.class) public static class BasicTests extends SharedTestBase implements Serializable { @Test @Category(ValidatesRunner.class) @@ -615,7 +615,7 @@ public void testLargeKeys100MB() throws Exception { } /** Tests validating GroupByKey behaviors with windowing. */ - @RunWith(JUnit4.class) + @RunWith(BeamParallelJunit4Runner.class) public static class WindowTests extends SharedTestBase { @Test @Category(ValidatesRunner.class) diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupIntoBatchesTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupIntoBatchesTest.java index 832eb03f05d1..92cb006c6f96 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupIntoBatchesTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupIntoBatchesTest.java @@ -29,6 +29,7 @@ import org.apache.beam.sdk.coders.IterableCoder; import org.apache.beam.sdk.coders.KvCoder; import org.apache.beam.sdk.coders.StringUtf8Coder; +import org.apache.beam.sdk.testing.BeamParallelJunit4Runner; import org.apache.beam.sdk.testing.NeedsRunner; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; @@ -66,12 +67,11 @@ import org.junit.experimental.categories.Category; import org.junit.rules.Timeout; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** Test Class for {@link GroupIntoBatches}. */ -@RunWith(JUnit4.class) +@RunWith(BeamParallelJunit4Runner.class) public class GroupIntoBatchesTest implements Serializable { private static final int BATCH_SIZE = 5; diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ImpulseTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ImpulseTest.java index 2039448770c1..31033ee9af90 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ImpulseTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ImpulseTest.java @@ -18,6 +18,7 @@ package org.apache.beam.sdk.transforms; import java.util.Arrays; +import org.apache.beam.sdk.testing.BeamParallelJunit4Runner; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; import org.apache.beam.sdk.testing.UsesImpulse; @@ -28,10 +29,9 @@ import org.junit.Test; import org.junit.experimental.categories.Category; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; /** Tests for Impulse. */ -@RunWith(JUnit4.class) +@RunWith(BeamParallelJunit4Runner.class) public class ImpulseTest { @Rule public transient TestPipeline p = TestPipeline.create(); diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/KeysTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/KeysTest.java index f7727d6d1170..40d1ebb0596f 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/KeysTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/KeysTest.java @@ -21,6 +21,7 @@ import org.apache.beam.sdk.coders.BigEndianIntegerCoder; import org.apache.beam.sdk.coders.KvCoder; import org.apache.beam.sdk.coders.StringUtf8Coder; +import org.apache.beam.sdk.testing.BeamParallelJunit4Runner; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; import org.apache.beam.sdk.testing.ValidatesRunner; @@ -30,10 +31,9 @@ import org.junit.Test; import org.junit.experimental.categories.Category; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; /** Tests for Keys transform. */ -@RunWith(JUnit4.class) +@RunWith(BeamParallelJunit4Runner.class) public class KeysTest { @SuppressWarnings({ "rawtypes", // TODO(https://github.com/apache/beam/issues/20447) diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/KvSwapTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/KvSwapTest.java index 5746d15d53cb..3851421e3868 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/KvSwapTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/KvSwapTest.java @@ -22,6 +22,7 @@ import org.apache.beam.sdk.coders.KvCoder; import org.apache.beam.sdk.coders.NullableCoder; import org.apache.beam.sdk.coders.StringUtf8Coder; +import org.apache.beam.sdk.testing.BeamParallelJunit4Runner; import org.apache.beam.sdk.testing.NeedsRunner; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; @@ -32,10 +33,9 @@ import org.junit.Test; import org.junit.experimental.categories.Category; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; /** Tests for KvSwap transform. */ -@RunWith(JUnit4.class) +@RunWith(BeamParallelJunit4Runner.class) @SuppressWarnings({ "rawtypes", // TODO(https://github.com/apache/beam/issues/20447) "unchecked", diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/MapElementsTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/MapElementsTest.java index dbbcb0e20d21..7a0ec6d03e18 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/MapElementsTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/MapElementsTest.java @@ -31,6 +31,7 @@ import java.io.Serializable; import java.util.Map; import java.util.Set; +import org.apache.beam.sdk.testing.BeamParallelJunit4Runner; import org.apache.beam.sdk.testing.NeedsRunner; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; @@ -52,10 +53,9 @@ import org.junit.rules.ExpectedException; import org.junit.rules.Timeout; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; /** Tests for {@link MapElements}. */ -@RunWith(JUnit4.class) +@RunWith(BeamParallelJunit4Runner.class) public class MapElementsTest implements Serializable { @Rule public final transient TestPipeline pipeline = TestPipeline.create(); diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/MapViewTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/MapViewTest.java index 005feda63ab9..acb78f9e68f7 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/MapViewTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/MapViewTest.java @@ -37,6 +37,7 @@ import org.apache.beam.sdk.coders.NullableCoder; import org.apache.beam.sdk.coders.StringUtf8Coder; import org.apache.beam.sdk.coders.VarIntCoder; +import org.apache.beam.sdk.testing.BeamParallelJunit4Runner; import org.apache.beam.sdk.testing.NeedsRunner; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; @@ -57,13 +58,12 @@ import org.junit.rules.ExpectedException; import org.junit.rules.Timeout; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; /** * Tests for (multi)map {@link View}. See also {@link ParDoTest}, which provides additional coverage * since views can only be observed via {@link ParDo}. */ -@RunWith(JUnit4.class) +@RunWith(BeamParallelJunit4Runner.class) @Category(UsesSideInputs.class) public class MapViewTest implements Serializable { // This test is Serializable, just so that it's easy to have diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoLifecycleTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoLifecycleTest.java index 21b4f64f9247..f384330b3868 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoLifecycleTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoLifecycleTest.java @@ -44,6 +44,7 @@ import org.apache.beam.sdk.state.StateSpec; import org.apache.beam.sdk.state.StateSpecs; import org.apache.beam.sdk.state.ValueState; +import org.apache.beam.sdk.testing.BeamParallelJunit4Runner; import org.apache.beam.sdk.testing.TestPipeline; import org.apache.beam.sdk.testing.UsesParDoLifecycle; import org.apache.beam.sdk.testing.UsesStatefulParDo; @@ -58,10 +59,9 @@ import org.junit.Test; import org.junit.experimental.categories.Category; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; /** Tests that {@link ParDo} exercises {@link DoFn} methods in the appropriate sequence. */ -@RunWith(JUnit4.class) +@RunWith(BeamParallelJunit4Runner.class) @SuppressWarnings({ "rawtypes", // TODO(https://github.com/apache/beam/issues/20447) // TODO(https://github.com/apache/beam/issues/21230): Remove when new version of @@ -176,6 +176,7 @@ private static class CallSequenceEnforcingStatefulFn } @Test + @BeamParallelJunit4Runner.SerialTest @Category({ValidatesRunner.class, UsesParDoLifecycle.class}) public void testTeardownCalledAfterExceptionInSetup() { ExceptionThrowingFn fn = new ExceptionThrowingFn(MethodForException.SETUP); @@ -189,6 +190,7 @@ public void testTeardownCalledAfterExceptionInSetup() { } @Test + @BeamParallelJunit4Runner.SerialTest @Category({ValidatesRunner.class, UsesParDoLifecycle.class}) public void testTeardownCalledAfterExceptionInStartBundle() { ExceptionThrowingFn fn = new ExceptionThrowingFn(MethodForException.START_BUNDLE); @@ -202,6 +204,7 @@ public void testTeardownCalledAfterExceptionInStartBundle() { } @Test + @BeamParallelJunit4Runner.SerialTest @Category({ValidatesRunner.class, UsesParDoLifecycle.class}) public void testTeardownCalledAfterExceptionInProcessElement() { ExceptionThrowingFn fn = new ExceptionThrowingFn(MethodForException.PROCESS_ELEMENT); @@ -216,6 +219,7 @@ public void testTeardownCalledAfterExceptionInProcessElement() { } @Test + @BeamParallelJunit4Runner.SerialTest @Category({ValidatesRunner.class, UsesParDoLifecycle.class}) public void testTeardownCalledAfterExceptionInFinishBundle() { ExceptionThrowingFn fn = new ExceptionThrowingFn(MethodForException.FINISH_BUNDLE); @@ -234,6 +238,7 @@ public void testTeardownCalledAfterExceptionInFinishBundle() { } @Test + @BeamParallelJunit4Runner.SerialTest @Category({ValidatesRunner.class, UsesStatefulParDo.class, UsesParDoLifecycle.class}) public void testTeardownCalledAfterExceptionInSetupStateful() { ExceptionThrowingFn fn = new ExceptionThrowingStatefulFn(MethodForException.SETUP); @@ -247,6 +252,7 @@ public void testTeardownCalledAfterExceptionInSetupStateful() { } @Test + @BeamParallelJunit4Runner.SerialTest @Category({ValidatesRunner.class, UsesStatefulParDo.class, UsesParDoLifecycle.class}) public void testTeardownCalledAfterExceptionInStartBundleStateful() { ExceptionThrowingFn fn = new ExceptionThrowingStatefulFn(MethodForException.START_BUNDLE); @@ -260,6 +266,7 @@ public void testTeardownCalledAfterExceptionInStartBundleStateful() { } @Test + @BeamParallelJunit4Runner.SerialTest @Category({ValidatesRunner.class, UsesStatefulParDo.class, UsesParDoLifecycle.class}) public void testTeardownCalledAfterExceptionInProcessElementStateful() { ExceptionThrowingFn fn = new ExceptionThrowingStatefulFn(MethodForException.PROCESS_ELEMENT); @@ -274,6 +281,7 @@ public void testTeardownCalledAfterExceptionInProcessElementStateful() { } @Test + @BeamParallelJunit4Runner.SerialTest @Category({ValidatesRunner.class, UsesStatefulParDo.class, UsesParDoLifecycle.class}) public void testTeardownCalledAfterExceptionInFinishBundleStateful() { ExceptionThrowingFn fn = new ExceptionThrowingStatefulFn(MethodForException.FINISH_BUNDLE); diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoSchemaTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoSchemaTest.java index 07538a40d1ff..42d92bfc7cf9 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoSchemaTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoSchemaTest.java @@ -43,6 +43,7 @@ import org.apache.beam.sdk.state.SetState; import org.apache.beam.sdk.state.StateSpec; import org.apache.beam.sdk.state.StateSpecs; +import org.apache.beam.sdk.testing.BeamParallelJunit4Runner; import org.apache.beam.sdk.testing.NeedsRunner; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; @@ -66,10 +67,9 @@ import org.junit.rules.ExpectedException; import org.junit.rules.Timeout; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; /** Test {@link Schema} support. */ -@RunWith(JUnit4.class) +@RunWith(BeamParallelJunit4Runner.class) @Category(UsesSchema.class) // TODO(https://github.com/apache/beam/issues/21230): Remove when new version of errorprone is // released (2.11.0) diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java index 7eb704b2bbfb..5fc04afbc0fe 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java @@ -53,13 +53,13 @@ import java.util.Arrays; import java.util.Collections; import java.util.Comparator; -import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Map.Entry; import java.util.Set; import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.IntFunction; import java.util.stream.Collectors; @@ -98,6 +98,7 @@ import org.apache.beam.sdk.state.TimerSpec; import org.apache.beam.sdk.state.TimerSpecs; import org.apache.beam.sdk.state.ValueState; +import org.apache.beam.sdk.testing.BeamParallelJunit4Runner; import org.apache.beam.sdk.testing.NeedsRunner; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; @@ -176,7 +177,6 @@ import org.junit.rules.ExpectedException; import org.junit.rules.Timeout; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; /** Tests for ParDo. */ @SuppressWarnings({ @@ -391,7 +391,7 @@ public void processElement(@Element Integer element, OutputReceiver r) } /** Tests for basic {@link ParDo} scenarios. */ - @RunWith(JUnit4.class) + @RunWith(BeamParallelJunit4Runner.class) public static class BasicTests extends SharedTestBase implements Serializable { @Test @Category(ValidatesRunner.class) @@ -682,7 +682,7 @@ public void process(OutputReceiver r) { } /** Tests to validate behaviors around multiple inputs or outputs. */ - @RunWith(JUnit4.class) + @RunWith(BeamParallelJunit4Runner.class) public static class MultipleInputsAndOutputTests extends SharedTestBase implements Serializable { @Test @Category(ValidatesRunner.class) @@ -1591,13 +1591,13 @@ public void populateDisplayData(DisplayData.Builder builder) { } } - @RunWith(JUnit4.class) + @RunWith(BeamParallelJunit4Runner.class) public static class BundleFinalizationTests extends SharedTestBase implements Serializable { private abstract static class BundleFinalizingDoFn extends DoFn, String> { private static final long MAX_ATTEMPTS = 100; // We use the UUID to uniquely identify this DoFn in case this test is run with // other tests in the same JVM. - private static final Map WAS_FINALIZED = new HashMap(); + private static final Map WAS_FINALIZED = new ConcurrentHashMap<>(); private final UUID uuid = UUID.randomUUID(); public void testFinalization(BundleFinalizer bundleFinalizer, OutputReceiver output) @@ -1746,7 +1746,7 @@ public void testBundleFinalizationWithSideInputs() { } /** Tests for ParDo lifecycle methods. */ - @RunWith(JUnit4.class) + @RunWith(BeamParallelJunit4Runner.class) public static class LifecycleTests extends SharedTestBase implements Serializable { @Test @Category(ValidatesRunner.class) @@ -1821,7 +1821,7 @@ public void finishBundle(FinishBundleContext c) { } /** Tests to validate output timestamps. */ - @RunWith(JUnit4.class) + @RunWith(BeamParallelJunit4Runner.class) public static class TimestampTests extends SharedTestBase implements Serializable { static final String TIMER_ELEMENT = "timer"; @@ -2278,7 +2278,7 @@ public void testOnWindowTimestampSkew() { } /** Tests to validate ParDo state. */ - @RunWith(JUnit4.class) + @RunWith(BeamParallelJunit4Runner.class) public static class StateTests extends SharedTestBase implements Serializable { @Test @Category({ValidatesRunner.class, UsesStatefulParDo.class}) @@ -4124,7 +4124,7 @@ public void process( } /** Tests for state coder inference behaviors. */ - @RunWith(JUnit4.class) + @RunWith(BeamParallelJunit4Runner.class) public static class StateCoderInferenceTests extends SharedTestBase implements Serializable { @Test @Category({ValidatesRunner.class, UsesStatefulParDo.class}) @@ -4649,7 +4649,7 @@ public void processElement( } /** Tests to validate ParDo timers. */ - @RunWith(JUnit4.class) + @RunWith(BeamParallelJunit4Runner.class) public static class TimerTests extends SharedTestBase implements Serializable { @Test @@ -6532,7 +6532,7 @@ public void onTimer(OutputReceiver r) { } /** Tests validating Timer coder inference behaviors. */ - @RunWith(JUnit4.class) + @RunWith(BeamParallelJunit4Runner.class) public static class TimerCoderInferenceTests extends SharedTestBase implements Serializable { @Test @Category({ValidatesRunner.class, UsesStatefulParDo.class}) @@ -6904,7 +6904,7 @@ public void onTimer(OutputReceiver r, @TimerId("timer") Timer timer) { } /** Tests to validate ParDo timerFamily. */ - @RunWith(JUnit4.class) + @RunWith(BeamParallelJunit4Runner.class) public static class TimerFamilyTests extends SharedTestBase implements Serializable { @Test @@ -7130,7 +7130,7 @@ public void onTimer(TimeDomain timeDomain, OutputReceiver r) { } /** Tests to validate Key in OnTimer. */ - @RunWith(JUnit4.class) + @RunWith(BeamParallelJunit4Runner.class) public static class KeyTests extends SharedTestBase implements Serializable { @Test @@ -7264,7 +7264,7 @@ public void onTimer(@Key Integer key, OutputReceiver r) {} } /** Tests to validate OnWindowExpiration. */ - @RunWith(JUnit4.class) + @RunWith(BeamParallelJunit4Runner.class) public static class OnWindowExpirationTests extends SharedTestBase implements Serializable { @Test @@ -7392,7 +7392,7 @@ public void onWindowExpiration( } /** Tests to validate SchemaInformation. */ - @RunWith(JUnit4.class) + @RunWith(BeamParallelJunit4Runner.class) public static class SchemaInformationTests extends SharedTestBase implements Serializable { private static final Schema TEST_SCHEMA = @@ -7514,7 +7514,7 @@ public void processElement(OutputReceiver r) { } } - @RunWith(JUnit4.class) + @RunWith(BeamParallelJunit4Runner.class) public static class BundleInvariantsTests extends SharedTestBase implements Serializable { @Test diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/PerKeyOrderingTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/PerKeyOrderingTest.java index b023127cc522..6bcdb7218588 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/PerKeyOrderingTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/PerKeyOrderingTest.java @@ -28,6 +28,7 @@ import org.apache.beam.sdk.state.StateSpec; import org.apache.beam.sdk.state.StateSpecs; import org.apache.beam.sdk.state.ValueState; +import org.apache.beam.sdk.testing.BeamParallelJunit4Runner; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; import org.apache.beam.sdk.testing.UsesPerKeyOrderInBundle; @@ -44,12 +45,11 @@ import org.junit.Test; import org.junit.experimental.categories.Category; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; @SuppressWarnings( "unused") // TODO(https://github.com/apache/beam/issues/21230): Remove when new version of // errorprone is released (2.11.0) -@RunWith(JUnit4.class) +@RunWith(BeamParallelJunit4Runner.class) public class PerKeyOrderingTest implements Serializable { @Rule public final transient TestPipeline pipeline = TestPipeline.create(); diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/RedistributeTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/RedistributeTest.java index ea46ffec4496..d0ca757cd620 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/RedistributeTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/RedistributeTest.java @@ -33,6 +33,7 @@ import org.apache.beam.sdk.coders.StringUtf8Coder; import org.apache.beam.sdk.coders.VarIntCoder; import org.apache.beam.sdk.coders.VarLongCoder; +import org.apache.beam.sdk.testing.BeamParallelJunit4Runner; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; import org.apache.beam.sdk.testing.TestStream; @@ -61,10 +62,9 @@ import org.junit.Test; import org.junit.experimental.categories.Category; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; /** Tests for {@link Redistribute}. */ -@RunWith(JUnit4.class) +@RunWith(BeamParallelJunit4Runner.class) public class RedistributeTest implements Serializable { private static final ImmutableList> ARBITRARY_KVS = diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ReifyTimestampsTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ReifyTimestampsTest.java index c76fb72325f7..d488c6fc608f 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ReifyTimestampsTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ReifyTimestampsTest.java @@ -21,6 +21,7 @@ import static org.hamcrest.Matchers.equalTo; import java.io.Serializable; +import org.apache.beam.sdk.testing.BeamParallelJunit4Runner; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; import org.apache.beam.sdk.testing.ValidatesRunner; @@ -32,10 +33,9 @@ import org.junit.Test; import org.junit.experimental.categories.Category; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; /** Tests for {@link ReifyTimestamps}. */ -@RunWith(JUnit4.class) +@RunWith(BeamParallelJunit4Runner.class) public class ReifyTimestampsTest implements Serializable { @Rule public transient TestPipeline pipeline = TestPipeline.create(); diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ReshuffleTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ReshuffleTest.java index 10f65dfce68d..40d004f50a47 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ReshuffleTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ReshuffleTest.java @@ -36,6 +36,7 @@ import org.apache.beam.sdk.coders.VarLongCoder; import org.apache.beam.sdk.options.StreamingOptions; import org.apache.beam.sdk.runners.TransformHierarchy; +import org.apache.beam.sdk.testing.BeamParallelJunit4Runner; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; import org.apache.beam.sdk.testing.TestStream; @@ -63,10 +64,9 @@ import org.junit.Test; import org.junit.experimental.categories.Category; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; /** Tests for {@link Reshuffle}. */ -@RunWith(JUnit4.class) +@RunWith(BeamParallelJunit4Runner.class) public class ReshuffleTest implements Serializable { private static final ImmutableList> ARBITRARY_KVS = diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/SplittableDoFnTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/SplittableDoFnTest.java index 80d8728aa01b..27affb20624e 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/SplittableDoFnTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/SplittableDoFnTest.java @@ -41,6 +41,7 @@ import org.apache.beam.sdk.coders.KvCoder; import org.apache.beam.sdk.coders.StringUtf8Coder; import org.apache.beam.sdk.io.range.OffsetRange; +import org.apache.beam.sdk.testing.BeamParallelJunit4Runner; import org.apache.beam.sdk.testing.NeedsRunner; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; @@ -85,12 +86,11 @@ import org.junit.experimental.categories.Category; import org.junit.rules.Timeout; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; /** * Tests for splittable {@link DoFn} behavior. */ -@RunWith(JUnit4.class) +@RunWith(BeamParallelJunit4Runner.class) @SuppressWarnings({ "rawtypes", // TODO(https://github.com/apache/beam/issues/20447) }) diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ViewTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ViewTest.java index 0301f4568f88..02d91c488548 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ViewTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ViewTest.java @@ -35,6 +35,7 @@ import org.apache.beam.sdk.coders.VarLongCoder; import org.apache.beam.sdk.coders.VoidCoder; import org.apache.beam.sdk.io.GenerateSequence; +import org.apache.beam.sdk.testing.BeamParallelJunit4Runner; import org.apache.beam.sdk.testing.NeedsRunner; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; @@ -65,13 +66,12 @@ import org.junit.rules.ExpectedException; import org.junit.rules.Timeout; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; /** * Tests for {@link View}. See also {@link ParDoTest}, which provides additional coverage since * views can only be observed via {@link ParDo}. */ -@RunWith(JUnit4.class) +@RunWith(BeamParallelJunit4Runner.class) @Category(UsesSideInputs.class) public class ViewTest implements Serializable { // This test is Serializable, just so that it's easy to have diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/WithTimestampsTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/WithTimestampsTest.java index 5b8eb6c256e8..cbfb31ed8d24 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/WithTimestampsTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/WithTimestampsTest.java @@ -21,6 +21,7 @@ import java.io.Serializable; import org.apache.beam.sdk.Pipeline.PipelineExecutionException; +import org.apache.beam.sdk.testing.BeamParallelJunit4Runner; import org.apache.beam.sdk.testing.NeedsRunner; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; @@ -35,10 +36,9 @@ import org.junit.rules.ExpectedException; import org.junit.rules.Timeout; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; /** Tests for {@link WithTimestamps}. */ -@RunWith(JUnit4.class) +@RunWith(BeamParallelJunit4Runner.class) public class WithTimestampsTest implements Serializable { @Rule public final transient TestPipeline p = TestPipeline.create(); diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/join/CoGroupByKeyTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/join/CoGroupByKeyTest.java index e89210736755..5788195c3026 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/join/CoGroupByKeyTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/join/CoGroupByKeyTest.java @@ -30,6 +30,7 @@ import org.apache.beam.sdk.coders.KvCoder; import org.apache.beam.sdk.coders.StringUtf8Coder; import org.apache.beam.sdk.coders.VarIntCoder; +import org.apache.beam.sdk.testing.BeamParallelJunit4Runner; import org.apache.beam.sdk.testing.NeedsRunner; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; @@ -52,10 +53,9 @@ import org.junit.Test; import org.junit.experimental.categories.Category; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; /** Tests for CoGroupByKeyTest. Implements Serializable for anonymous DoFns. */ -@RunWith(JUnit4.class) +@RunWith(BeamParallelJunit4Runner.class) public class CoGroupByKeyTest implements Serializable { /** diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/WindowTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/WindowTest.java index b0b82430a440..16f939e26b67 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/WindowTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/WindowTest.java @@ -51,6 +51,7 @@ import org.apache.beam.sdk.coders.VarIntCoder; import org.apache.beam.sdk.io.GenerateSequence; import org.apache.beam.sdk.runners.TransformHierarchy; +import org.apache.beam.sdk.testing.BeamParallelJunit4Runner; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; import org.apache.beam.sdk.testing.UsesCustomWindowMerging; @@ -82,11 +83,10 @@ import org.junit.experimental.categories.Category; import org.junit.rules.ExpectedException; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; import org.mockito.Mockito; /** Tests for {@link Window}. */ -@RunWith(JUnit4.class) +@RunWith(BeamParallelJunit4Runner.class) public class WindowTest implements Serializable { @Rule diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/WindowingTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/WindowingTest.java index c1c3e86f5711..862591ecc54f 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/WindowingTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/WindowingTest.java @@ -24,6 +24,7 @@ import java.util.List; import org.apache.beam.sdk.coders.StringUtf8Coder; import org.apache.beam.sdk.io.TextIO; +import org.apache.beam.sdk.testing.BeamParallelJunit4Runner; import org.apache.beam.sdk.testing.NeedsRunner; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; @@ -46,10 +47,9 @@ import org.junit.experimental.categories.Category; import org.junit.rules.TemporaryFolder; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; /** Unit tests for bucketing. */ -@RunWith(JUnit4.class) +@RunWith(BeamParallelJunit4Runner.class) public class WindowingTest implements Serializable { @Rule public final transient TestPipeline p = TestPipeline.create(); diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/values/PCollectionRowTupleTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/values/PCollectionRowTupleTest.java index 60b5ef0267b1..fee843d9d72e 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/values/PCollectionRowTupleTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/values/PCollectionRowTupleTest.java @@ -34,6 +34,7 @@ import org.apache.beam.sdk.schemas.Schema; import org.apache.beam.sdk.schemas.Schema.Field; import org.apache.beam.sdk.schemas.Schema.FieldType; +import org.apache.beam.sdk.testing.BeamParallelJunit4Runner; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; import org.apache.beam.sdk.testing.ValidatesRunner; @@ -46,10 +47,9 @@ import org.junit.Test; import org.junit.experimental.categories.Category; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; /** Unit tests for {@link PCollectionRowTuple}. */ -@RunWith(JUnit4.class) +@RunWith(BeamParallelJunit4Runner.class) public final class PCollectionRowTupleTest implements Serializable { public static final Schema INT_SCHEMA = Schema.of(Field.of("int", FieldType.INT32)); diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/values/PCollectionTupleTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/values/PCollectionTupleTest.java index 1a2b56749a91..3631f126366d 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/values/PCollectionTupleTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/values/PCollectionTupleTest.java @@ -32,6 +32,7 @@ import org.apache.beam.sdk.Pipeline; import org.apache.beam.sdk.coders.VarIntCoder; import org.apache.beam.sdk.io.GenerateSequence; +import org.apache.beam.sdk.testing.BeamParallelJunit4Runner; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; import org.apache.beam.sdk.testing.ValidatesRunner; @@ -46,10 +47,9 @@ import org.junit.Test; import org.junit.experimental.categories.Category; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; /** Unit tests for {@link PCollectionTuple}. */ -@RunWith(JUnit4.class) +@RunWith(BeamParallelJunit4Runner.class) public final class PCollectionTupleTest implements Serializable { @Rule diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/values/PDoneTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/values/PDoneTest.java index bb9be49396d5..9f7771ef2dc4 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/values/PDoneTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/values/PDoneTest.java @@ -21,6 +21,7 @@ import java.io.File; import org.apache.beam.sdk.io.TextIO; +import org.apache.beam.sdk.testing.BeamParallelJunit4Runner; import org.apache.beam.sdk.testing.NeedsRunner; import org.apache.beam.sdk.testing.TestPipeline; import org.apache.beam.sdk.testing.ValidatesRunner; @@ -32,10 +33,9 @@ import org.junit.experimental.categories.Category; import org.junit.rules.TemporaryFolder; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; /** Tests for PDone. */ -@RunWith(JUnit4.class) +@RunWith(BeamParallelJunit4Runner.class) public class PDoneTest { @Rule public final TestPipeline p = TestPipeline.create();