This is an automated email from the ASF dual-hosted git repository.
reuvenlax pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 43bd833c88a Merge pull request #40292 from
reuvenlax/merge_validates_runner
43bd833c88a is described below
commit 43bd833c88a1149aa025195218544479797e2c39
Author: Reuven Lax <[email protected]>
AuthorDate: Wed Sep 30 15:29:10 2026 -0700
Merge pull request #40292 from reuvenlax/merge_validates_runner
Improve ValidatesRunner performance. Parallelize runs
---
...m_PostCommit_Java_ValidatesRunner_Dataflow.json | 2 +-
runners/google-cloud-dataflow-java/build.gradle | 25 ++-
.../beam/sdk/testing/BeamParallelJunit4Runner.java | 189 +++++++++++++++++++
.../ClasspathScanningResourcesDetector.java | 1 -
.../java/org/apache/beam/sdk/PipelineTest.java | 7 +-
.../apache/beam/sdk/SdkHarnessEnvironmentTest.java | 4 +-
.../org/apache/beam/sdk/io/CountingSourceTest.java | 4 +-
.../org/apache/beam/sdk/metrics/MetricsTest.java | 9 +-
.../sdk/testing/BeamParallelJunit4RunnerTest.java | 203 +++++++++++++++++++++
.../org/apache/beam/sdk/testing/PAssertTest.java | 3 +-
.../apache/beam/sdk/testing/TestStreamTest.java | 3 +-
.../apache/beam/sdk/transforms/CombineFnsTest.java | 9 +-
.../apache/beam/sdk/transforms/CombineTest.java | 10 +-
.../org/apache/beam/sdk/transforms/CreateTest.java | 4 +-
.../apache/beam/sdk/transforms/FlattenTest.java | 4 +-
.../apache/beam/sdk/transforms/GroupByKeyTest.java | 6 +-
.../beam/sdk/transforms/GroupIntoBatchesTest.java | 4 +-
.../apache/beam/sdk/transforms/ImpulseTest.java | 4 +-
.../org/apache/beam/sdk/transforms/KeysTest.java | 4 +-
.../org/apache/beam/sdk/transforms/KvSwapTest.java | 4 +-
.../beam/sdk/transforms/MapElementsTest.java | 4 +-
.../apache/beam/sdk/transforms/MapViewTest.java | 4 +-
.../beam/sdk/transforms/ParDoLifecycleTest.java | 12 +-
.../beam/sdk/transforms/ParDoSchemaTest.java | 4 +-
.../org/apache/beam/sdk/transforms/ParDoTest.java | 34 ++--
.../beam/sdk/transforms/PerKeyOrderingTest.java | 4 +-
.../beam/sdk/transforms/RedistributeTest.java | 4 +-
.../beam/sdk/transforms/ReifyTimestampsTest.java | 4 +-
.../apache/beam/sdk/transforms/ReshuffleTest.java | 4 +-
.../beam/sdk/transforms/SplittableDoFnTest.java | 4 +-
.../org/apache/beam/sdk/transforms/ViewTest.java | 4 +-
.../beam/sdk/transforms/WithTimestampsTest.java | 4 +-
.../beam/sdk/transforms/join/CoGroupByKeyTest.java | 4 +-
.../beam/sdk/transforms/windowing/WindowTest.java | 4 +-
.../sdk/transforms/windowing/WindowingTest.java | 4 +-
.../beam/sdk/values/PCollectionRowTupleTest.java | 4 +-
.../beam/sdk/values/PCollectionTupleTest.java | 4 +-
.../java/org/apache/beam/sdk/values/PDoneTest.java | 4 +-
38 files changed, 515 insertions(+), 94 deletions(-)
diff --git
a/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Dataflow.json
b/.github/trigger_files/beam_PostCommit_Java_ValidatesRunner_Dataflow.json
index 8144784f5f0..97b7503f544 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 20c09704d41..02d985c5fe4 100644
--- a/runners/google-cloud-dataflow-java/build.gradle
+++ b/runners/google-cloud-dataflow-java/build.gradle
@@ -239,16 +239,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 ->
@@ -261,10 +276,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)
@@ -295,10 +312,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 00000000000..bcaead005ad
--- /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:
+ *
+ * <ul>
+ * <li>{@code -Dbeam.validatesRunner.parallelThreads=N}: number of
concurrent threads per test
+ * worker JVM for {@code @Test} methods annotated with
+ * {@code @Category(ValidatesRunner.class)} (defaults to {@code
beam.test.parallelThreads}).
+ * <li>{@code -Dbeam.test.parallelThreads=N}: number of concurrent threads
per test worker JVM for
+ * all other {@code @Test} methods (defaults to {@code 1}, i.e.,
sequential execution).
+ * </ul>
+ *
+ * <p>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<String, ExecutorService> EXECUTORS =
+ new ConcurrentHashMap<>();
+
+ private static final ConcurrentHashMap<Class<?>, Boolean> CLASS_SERIAL_CACHE
=
+ new ConcurrentHashMap<>();
+
+ private CompletableFuture<Void> 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 5a5c6b32527..c1d815de6ab 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 class ClasspathScanningResourcesDetector implements
PipelineResourcesDete
if (snapshot != null && snapshot.matches(classLoader,
currentJavaClassPath)) {
return new ArrayList<>(snapshot.files);
}
-
List<File> 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 4a4ef1dca6a..a5c74b43ab3 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.PTransformMatcher;
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.experimental.categories.Category;
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 class PipelineTest {
}
@Test
+ @BeamParallelJunit4Runner.SerialTest
public void testStableUniqueNameOff() {
pipeline.enableAbandonedNodeEnforcement(false);
@@ -285,6 +286,7 @@ public class PipelineTest {
}
@Test
+ @BeamParallelJunit4Runner.SerialTest
public void testStableUniqueNameWarning() {
pipeline.enableAbandonedNodeEnforcement(false);
@@ -297,6 +299,7 @@ public class PipelineTest {
}
@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 f258529bee2..a04a123eae1 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.coders.StringUtf8Coder;
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.Rule;
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 337462340df..b8542945d50 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.CounterMark;
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.Rule;
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<Long> 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 fbf0ab17282..b49ca6c36de 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
@@ -43,6 +43,7 @@ import org.apache.beam.sdk.io.GenerateSequence;
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;
@@ -69,7 +70,6 @@ import org.junit.Test;
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}. */
@@ -191,7 +191,7 @@ public class MetricsTest implements Serializable {
}
/** Tests validating basic metric scenarios. */
- @RunWith(JUnit4.class)
+ @RunWith(BeamParallelJunit4Runner.class)
public static class BasicTests extends SharedTestBase {
@Test
public void testDistributionWithoutContainer() {
@@ -274,6 +274,7 @@ public class MetricsTest implements Serializable {
}
@Test
+ @BeamParallelJunit4Runner.SerialTest
public void testMetricsFlag() {
Metrics.resetDefaultPipelineOptions();
assertFalse(Metrics.MetricsFlag.counterDisabled());
@@ -308,7 +309,7 @@ public class MetricsTest implements Serializable {
}
/** Tests for committed metrics. */
- @RunWith(JUnit4.class)
+ @RunWith(BeamParallelJunit4Runner.class)
public static class CommittedMetricTests extends SharedTestBase {
@Category({
ValidatesRunner.class,
@@ -445,7 +446,7 @@ public class MetricsTest implements Serializable {
}
/** 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 00000000000..1fd2b168bd0
--- /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<String> VR_THREADS = ConcurrentHashMap.newKeySet();
+ static final Set<String> 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 a02196bb2c0..bec3de5a624 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.Test;
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 00b6027f945..e9fec99da07 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.Test;
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 328bfba8f3c..2575729b885 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.io.Serializable;
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.CoderException;
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.Test;
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 class CombineFnsTest {
private static class DisplayDataCombineFn extends Combine.CombineFn<String,
String, String> {
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 993b84a528d..c888cadedf1 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.SerializableCoder;
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.Test;
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 @@ public class CombineTest implements Serializable {
}
/** 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 class CombineTest implements Serializable {
}
/** 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 class CombineTest implements Serializable {
}
/** 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 class CombineTest implements Serializable {
}
/** 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 795c18f8ca1..2dffdf63f6b 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.NestedValueProvider;
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.Test;
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 74fb2dc99c8..34154cb65eb 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.StringUtf8Coder;
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.experimental.categories.Category;
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 18541437d5f..13af0c4b356 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.TimeDomain;
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.Rule;
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 class GroupByKeyTest implements Serializable {
}
/** 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 class GroupByKeyTest implements Serializable {
}
/** 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 832eb03f05d..92cb006c6f9 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 java.util.stream.Collectors;
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.Test;
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 2039448770c..31033ee9af9 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.Rule;
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 f7727d6d117..40d1ebb0596 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 java.util.Arrays;
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.Rule;
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 5746d15d53c..3851421e386 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.BigEndianIntegerCoder;
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.Rule;
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 dbbcb0e20d2..7a0ec6d03e1 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 static org.junit.Assert.assertEquals;
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.experimental.categories.Category;
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 005feda63ab..acb78f9e68f 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.KvCoder;
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.experimental.categories.Category;
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 21b4f64f924..f384330b386 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 java.util.stream.Collectors;
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.Rule;
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 @@ public class ParDoLifecycleTest implements Serializable {
}
@Test
+ @BeamParallelJunit4Runner.SerialTest
@Category({ValidatesRunner.class, UsesParDoLifecycle.class})
public void testTeardownCalledAfterExceptionInSetup() {
ExceptionThrowingFn fn = new ExceptionThrowingFn(MethodForException.SETUP);
@@ -189,6 +190,7 @@ public class ParDoLifecycleTest implements Serializable {
}
@Test
+ @BeamParallelJunit4Runner.SerialTest
@Category({ValidatesRunner.class, UsesParDoLifecycle.class})
public void testTeardownCalledAfterExceptionInStartBundle() {
ExceptionThrowingFn fn = new
ExceptionThrowingFn(MethodForException.START_BUNDLE);
@@ -202,6 +204,7 @@ public class ParDoLifecycleTest implements Serializable {
}
@Test
+ @BeamParallelJunit4Runner.SerialTest
@Category({ValidatesRunner.class, UsesParDoLifecycle.class})
public void testTeardownCalledAfterExceptionInProcessElement() {
ExceptionThrowingFn fn = new
ExceptionThrowingFn(MethodForException.PROCESS_ELEMENT);
@@ -216,6 +219,7 @@ public class ParDoLifecycleTest implements Serializable {
}
@Test
+ @BeamParallelJunit4Runner.SerialTest
@Category({ValidatesRunner.class, UsesParDoLifecycle.class})
public void testTeardownCalledAfterExceptionInFinishBundle() {
ExceptionThrowingFn fn = new
ExceptionThrowingFn(MethodForException.FINISH_BUNDLE);
@@ -234,6 +238,7 @@ public class ParDoLifecycleTest implements Serializable {
}
@Test
+ @BeamParallelJunit4Runner.SerialTest
@Category({ValidatesRunner.class, UsesStatefulParDo.class,
UsesParDoLifecycle.class})
public void testTeardownCalledAfterExceptionInSetupStateful() {
ExceptionThrowingFn fn = new
ExceptionThrowingStatefulFn(MethodForException.SETUP);
@@ -247,6 +252,7 @@ public class ParDoLifecycleTest implements Serializable {
}
@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 class ParDoLifecycleTest implements Serializable {
}
@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 class ParDoLifecycleTest implements Serializable {
}
@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 07538a40d1f..42d92bfc7cf 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.MapState;
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.experimental.categories.Category;
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 7eb704b2bbf..5fc04afbc0f 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.ArrayList;
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.TimerMap;
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.experimental.categories.Category;
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 class ParDoTest implements Serializable {
}
/** 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 class ParDoTest implements Serializable {
}
/** 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 class ParDoTest implements Serializable {
}
}
- @RunWith(JUnit4.class)
+ @RunWith(BeamParallelJunit4Runner.class)
public static class BundleFinalizationTests extends SharedTestBase
implements Serializable {
private abstract static class BundleFinalizingDoFn extends DoFn<KV<String,
Long>, 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<UUID, AtomicBoolean> WAS_FINALIZED = new
HashMap();
+ private static final Map<UUID, AtomicBoolean> WAS_FINALIZED = new
ConcurrentHashMap<>();
private final UUID uuid = UUID.randomUUID();
public void testFinalization(BundleFinalizer bundleFinalizer,
OutputReceiver<String> output)
@@ -1746,7 +1746,7 @@ public class ParDoTest implements Serializable {
}
/** 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 class ParDoTest implements Serializable {
}
/** 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 class ParDoTest implements Serializable {
}
/** 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 class ParDoTest implements Serializable {
}
/** 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 class ParDoTest implements Serializable {
}
/** 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 class ParDoTest implements Serializable {
}
/** 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 class ParDoTest implements Serializable {
}
/** 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 class ParDoTest implements Serializable {
}
/** 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 class ParDoTest implements Serializable {
}
/** Tests to validate OnWindowExpiration. */
- @RunWith(JUnit4.class)
+ @RunWith(BeamParallelJunit4Runner.class)
public static class OnWindowExpirationTests extends SharedTestBase
implements Serializable {
@Test
@@ -7392,7 +7392,7 @@ public class ParDoTest implements Serializable {
}
/** 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 class ParDoTest implements Serializable {
}
}
- @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 b023127cc52..6bcdb721858 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.coders.VarLongCoder;
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.Rule;
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 ea46ffec449..d0ca757cd62 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.KvCoder;
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.Rule;
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<KV<String, Integer>> 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 c76fb72325f..d488c6fc608 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.MatcherAssert.assertThat;
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.Rule;
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 10f65dfce68..40d004f50a4 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.VarIntCoder;
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.Rule;
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<KV<String, Integer>> 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 80d8728aa01..27affb20624 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.BigEndianIntegerCoder;
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.Test;
import org.junit.experimental.categories.Category;
import org.junit.rules.Timeout;
import org.junit.runner.RunWith;
-import org.junit.runners.JUnit4;
/**
* Tests for <a href="https://s.apache.org/splittable-do-fn";>splittable</a>
{@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 0301f4568f8..02d91c48854 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.VarIntCoder;
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.internal.matchers.ThrowableMessageMatcher;
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 5b8eb6c256e..cbfb31ed8d2 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 static org.hamcrest.Matchers.isA;
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.experimental.categories.Category;
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 e8921073675..5788195c302 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.BigEndianIntegerCoder;
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.Rule;
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 b0b82430a44..16f939e26b6 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.StringUtf8Coder;
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.Test;
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 c1c3e86f571..862591ecc54 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.io.Serializable;
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.Test;
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 60b5ef0267b..fee843d9d72 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.coders.RowCoder;
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.Rule;
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 1a2b56749a9..3631f126366 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 java.util.Map.Entry;
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.Rule;
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 bb9be49396d..9f7771ef2dc 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 static org.apache.beam.sdk.TestUtils.LINES;
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.Test;
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();
