From f4d293bc342d3c181c8dcbcb9384c97ff291d6ca Mon Sep 17 00:00:00 2001 From: Reuven Lax Date: Mon, 28 Sep 2026 11:19:19 -0700 Subject: [PATCH 1/5] Add BeamParallelJunit4Runner for parallel within-class ValidatesRunner execution --- ...tCommit_Java_ValidatesRunner_Dataflow.json | 2 +- .../google-cloud-dataflow-java/build.gradle | 25 +- .../runners/dataflow/DataflowRunnerTest.java | 4 +- .../sdk/testing/BeamParallelJunit4Runner.java | 256 ++++++++++++++++++ .../ClasspathScanningResourcesDetector.java | 1 - .../org/apache/beam/sdk/PipelineTest.java | 4 +- .../beam/sdk/SdkHarnessEnvironmentTest.java | 4 +- .../beam/sdk/io/CountingSourceTest.java | 4 +- .../apache/beam/sdk/metrics/MetricsTest.java | 165 ++++++----- .../testing/BeamParallelJunit4RunnerTest.java | 204 ++++++++++++++ .../apache/beam/sdk/testing/PAssertTest.java | 3 +- .../beam/sdk/testing/TestStreamTest.java | 3 +- .../beam/sdk/transforms/CombineFnsTest.java | 4 +- .../beam/sdk/transforms/CombineTest.java | 10 +- .../beam/sdk/transforms/CreateTest.java | 4 +- .../beam/sdk/transforms/FlattenTest.java | 4 +- .../beam/sdk/transforms/GroupByKeyTest.java | 6 +- .../sdk/transforms/GroupIntoBatchesTest.java | 4 +- .../beam/sdk/transforms/ImpulseTest.java | 4 +- .../apache/beam/sdk/transforms/KeysTest.java | 4 +- .../beam/sdk/transforms/KvSwapTest.java | 4 +- .../beam/sdk/transforms/MapElementsTest.java | 4 +- .../beam/sdk/transforms/MapViewTest.java | 4 +- .../sdk/transforms/ParDoLifecycleTest.java | 12 +- .../beam/sdk/transforms/ParDoSchemaTest.java | 4 +- .../apache/beam/sdk/transforms/ParDoTest.java | 30 +- .../sdk/transforms/PerKeyOrderingTest.java | 4 +- .../beam/sdk/transforms/RedistributeTest.java | 4 +- .../sdk/transforms/ReifyTimestampsTest.java | 4 +- .../beam/sdk/transforms/ReshuffleTest.java | 4 +- .../sdk/transforms/SplittableDoFnTest.java | 4 +- .../apache/beam/sdk/transforms/ViewTest.java | 4 +- .../sdk/transforms/WithTimestampsTest.java | 4 +- .../sdk/transforms/join/CoGroupByKeyTest.java | 4 +- .../sdk/transforms/windowing/WindowTest.java | 4 +- .../transforms/windowing/WindowingTest.java | 4 +- .../sdk/values/PCollectionRowTupleTest.java | 4 +- .../beam/sdk/values/PCollectionTupleTest.java | 4 +- .../org/apache/beam/sdk/values/PDoneTest.java | 4 +- 39 files changed, 662 insertions(+), 163 deletions(-) create mode 100644 sdks/java/core/src/main/java/org/apache/beam/sdk/testing/BeamParallelJunit4Runner.java create mode 100644 sdks/java/core/src/test/java/org/apache/beam/sdk/testing/BeamParallelJunit4RunnerTest.java 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/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java index 549e4f782a39..6ee47109c082 100644 --- a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java +++ b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java @@ -142,6 +142,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.ExpectedLogs; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; @@ -205,7 +206,6 @@ import org.junit.rules.ExpectedException; import org.junit.rules.TemporaryFolder; import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; import org.mockito.ArgumentCaptor; import org.mockito.Mockito; @@ -214,7 +214,7 @@ * *

Implements {@link Serializable} because it is caught in closures. */ -@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/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..805383cef2f2 --- /dev/null +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/BeamParallelJunit4Runner.java @@ -0,0 +1,256 @@ +/* + * 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.io.IOException; +import java.io.InputStream; +import java.lang.annotation.Annotation; +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.beam.sdk.annotations.Internal; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableSet; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.io.ByteStreams; +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}, {@code @NotThreadSafe} (such as {@code + * javax.annotation.concurrent.NotThreadSafe} or {@code net.jcip.annotations.NotThreadSafe}), + * {@code @Serial}, or {@code @Isolated} 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 Set SERIAL_ANNOTATION_SIMPLE_NAMES = + ImmutableSet.of("NotThreadSafe", "SerialTest", "Serial", "Isolated"); + + private static final ConcurrentHashMap EXECUTORS = + new ConcurrentHashMap<>(); + + private static final ConcurrentHashMap, Boolean> CLASS_SERIAL_CACHE = + new ConcurrentHashMap<>(); + + private final List> pendingFutures = Collections.synchronizedList(new ArrayList<>()); + + 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() { + Throwable firstError = null; + List> snapshot; + synchronized (pendingFutures) { + snapshot = new ArrayList<>(pendingFutures); + pendingFutures.clear(); + } + for (Future future : snapshot) { + try { + future.get(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + if (firstError == null) { + firstError = e; + } + } catch (ExecutionException e) { + if (firstError == null) { + firstError = e.getCause() != null ? e.getCause() : e; + } + } + } + if (firstError instanceof RuntimeException) { + throw (RuntimeException) firstError; + } else if (firstError instanceof Error) { + throw (Error) firstError; + } else if (firstError != null) { + throw new RuntimeException(firstError); + } + } + + @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)) { + awaitPendingFutures(); + super.runChild(method, notifier); + return; + } + pendingFutures.add(executor.submit(() -> super.runChild(method, notifier))); + } + + 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); + if (vrThreads == defaultThreads) { + return getOrCreateExecutor("beam-test-worker", 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 hasSerialAnnotation(method.getAnnotations()); + } + + 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 hasSerialAnnotation(clazz.getAnnotations()) + || hasClassFileNotThreadSafeAnnotation(clazz) + || computeClassSerial(clazz.getSuperclass()) + || computeClassSerial(clazz.getEnclosingClass()); + } + + private static boolean hasSerialAnnotation(Annotation[] annotations) { + for (Annotation annotation : annotations) { + if (SERIAL_ANNOTATION_SIMPLE_NAMES.contains(annotation.annotationType().getSimpleName())) { + return true; + } + } + return false; + } + + /** + * Inspects the {@code .class} bytecode for {@code RetentionPolicy.CLASS} annotations such as + * {@code javax.annotation.concurrent.NotThreadSafe} or {@code net.jcip.annotations.NotThreadSafe} + * that are not retained for runtime reflection via {@link Class#getAnnotations()}. + */ + private static boolean hasClassFileNotThreadSafeAnnotation(Class clazz) { + String resourceName = clazz.getName().replace('.', '/') + ".class"; + ClassLoader classLoader = clazz.getClassLoader(); + try (InputStream in = + classLoader != null + ? classLoader.getResourceAsStream(resourceName) + : ClassLoader.getSystemResourceAsStream(resourceName)) { + if (in == null) { + return false; + } + String raw = new String(ByteStreams.toByteArray(in), StandardCharsets.ISO_8859_1); + return raw.contains("/NotThreadSafe;"); + } catch (IOException e) { + return false; + } + } +} 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..6ddd854126cc 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) }) 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..539905b321a0 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 @@ -34,6 +34,7 @@ import java.io.Serializable; import java.util.List; import java.util.NoSuchElementException; +import java.util.concurrent.ConcurrentHashMap; import org.apache.beam.sdk.PipelineResult; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.coders.VarIntCoder; @@ -42,6 +43,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 +70,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}. */ @@ -91,6 +92,9 @@ private static MetricQueryResults queryTestMetrics(PipelineResult result) { /** Shared test helpers and setup/teardown. */ public abstract static class SharedTestBase implements Serializable { + private static final ConcurrentHashMap CACHED_METRIC_PIPELINE_RESULTS = + new ConcurrentHashMap<>(); + @Rule public final transient ExpectedException thrown = ExpectedException.none(); @Rule public final transient TestPipeline pipeline = TestPipeline.create(); @@ -101,82 +105,93 @@ public void tearDown() { } protected PipelineResult runPipelineWithMetrics() { - final Counter count = Metrics.counter(MetricsTest.class, "count"); - StringSet sideinputs = Metrics.stringSet(MetricsTest.class, "sideinputs"); - final TupleTag output1 = new TupleTag() {}; - final TupleTag output2 = new TupleTag() {}; - pipeline - .apply(Create.of(5, 8, 13)) - .apply( - "MyStep1", - ParDo.of( - new DoFn() { - Distribution bundleDist = Metrics.distribution(MetricsTest.class, "bundle"); - - @StartBundle - public void startBundle() { - bundleDist.update(10L); - } - - @SuppressWarnings("unused") - @ProcessElement - public void processElement(ProcessContext c) { - Distribution values = Metrics.distribution(MetricsTest.class, "input"); - StringSet sources = Metrics.stringSet(MetricsTest.class, "sources"); - BoundedTrie boundedTrieSources = - Metrics.boundedTrie(MetricsTest.class, "boundedTrieSources"); - count.inc(); - values.update(c.element()); - - c.output(c.element()); - c.output(c.element()); - sources.add("gcs"); - sources.add("gcs"); // repeated should appear once - sources.add("gcs", "gcs"); // repeated should appear once - sideinputs.add("bigtable", "spanner"); - boundedTrieSources.add(ImmutableList.of("ab_source", "cd_source")); - boundedTrieSources.add(ImmutableList.of("ef_source")); - } - - @DoFn.FinishBundle - public void finishBundle() { - bundleDist.update(40L); - } - })) - .apply( - "MyStep2", - ParDo.of( - new DoFn() { - @SuppressWarnings("unused") - @ProcessElement - public void processElement(ProcessContext c) { - Distribution values = Metrics.distribution(MetricsTest.class, "input"); - Gauge gauge = Metrics.gauge(MetricsTest.class, "my-gauge"); - StringSet sinks = Metrics.stringSet(MetricsTest.class, "sinks"); - BoundedTrie boundedTrieSinks = - Metrics.boundedTrie(MetricsTest.class, "boundedTrieSinks"); - Integer element = c.element(); - count.inc(); - values.update(element); - gauge.set(12L); - c.output(element); - sinks.add("bq", "kafka", "kafka"); // repeated should appear once - sideinputs.add("bigtable", "sql"); - boundedTrieSinks.add(ImmutableList.of("ab_sink", "cd_sink")); - boundedTrieSinks.add(ImmutableList.of("ef_sink")); - c.output(output2, element); - } - }) - .withOutputTags(output1, TupleTagList.of(output2))); - PipelineResult result = pipeline.run(); - - result.waitUntilFinish(); - return result; + String cacheKey = + pipeline.getOptions().getRunner().getName() + + ":" + + System.getProperty(TestPipeline.PROPERTY_BEAM_TEST_PIPELINE_OPTIONS, ""); + synchronized (CACHED_METRIC_PIPELINE_RESULTS) { + PipelineResult cachedResult = CACHED_METRIC_PIPELINE_RESULTS.get(cacheKey); + if (cachedResult != null) { + return cachedResult; + } + final Counter count = Metrics.counter(MetricsTest.class, "count"); + StringSet sideinputs = Metrics.stringSet(MetricsTest.class, "sideinputs"); + final TupleTag output1 = new TupleTag() {}; + final TupleTag output2 = new TupleTag() {}; + pipeline + .apply(Create.of(5, 8, 13)) + .apply( + "MyStep1", + ParDo.of( + new DoFn() { + Distribution bundleDist = Metrics.distribution(MetricsTest.class, "bundle"); + + @StartBundle + public void startBundle() { + bundleDist.update(10L); + } + + @SuppressWarnings("unused") + @ProcessElement + public void processElement(ProcessContext c) { + Distribution values = Metrics.distribution(MetricsTest.class, "input"); + StringSet sources = Metrics.stringSet(MetricsTest.class, "sources"); + BoundedTrie boundedTrieSources = + Metrics.boundedTrie(MetricsTest.class, "boundedTrieSources"); + count.inc(); + values.update(c.element()); + + c.output(c.element()); + c.output(c.element()); + sources.add("gcs"); + sources.add("gcs"); // repeated should appear once + sources.add("gcs", "gcs"); // repeated should appear once + sideinputs.add("bigtable", "spanner"); + boundedTrieSources.add(ImmutableList.of("ab_source", "cd_source")); + boundedTrieSources.add(ImmutableList.of("ef_source")); + } + + @DoFn.FinishBundle + public void finishBundle() { + bundleDist.update(40L); + } + })) + .apply( + "MyStep2", + ParDo.of( + new DoFn() { + @SuppressWarnings("unused") + @ProcessElement + public void processElement(ProcessContext c) { + Distribution values = Metrics.distribution(MetricsTest.class, "input"); + Gauge gauge = Metrics.gauge(MetricsTest.class, "my-gauge"); + StringSet sinks = Metrics.stringSet(MetricsTest.class, "sinks"); + BoundedTrie boundedTrieSinks = + Metrics.boundedTrie(MetricsTest.class, "boundedTrieSinks"); + Integer element = c.element(); + count.inc(); + values.update(element); + gauge.set(12L); + c.output(element); + sinks.add("bq", "kafka", "kafka"); // repeated should appear once + sideinputs.add("bigtable", "sql"); + boundedTrieSinks.add(ImmutableList.of("ab_sink", "cd_sink")); + boundedTrieSinks.add(ImmutableList.of("ef_sink")); + c.output(output2, element); + } + }) + .withOutputTags(output1, TupleTagList.of(output2))); + PipelineResult result = pipeline.run(); + + result.waitUntilFinish(); + CACHED_METRIC_PIPELINE_RESULTS.put(cacheKey, result); + return result; + } } } /** Tests validating basic metric scenarios. */ - @RunWith(JUnit4.class) + @RunWith(BeamParallelJunit4Runner.class) public static class BasicTests extends SharedTestBase { @Test public void testDistributionWithoutContainer() { @@ -293,7 +308,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 +445,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..f5912e2de14c --- /dev/null +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/BeamParallelJunit4RunnerTest.java @@ -0,0 +1,204 @@ +/* + * 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 javax.annotation.concurrent.NotThreadSafe; +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(); + } + } + + @NotThreadSafe + @RunWith(BeamParallelJunit4Runner.class) + public static class SampleNotThreadSafeCases { + 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 testNotThreadSafeClassRunsSerially() { + System.setProperty(BeamParallelJunit4Runner.VALIDATES_RUNNER_THREADS_PROPERTY, "4"); + + IN_TEST_HARNESS.set(true); + try { + Result result = JUnitCore.runClasses(SampleNotThreadSafeCases.class); + assertEquals(0, result.getFailureCount()); + assertEquals(2, result.getRunCount()); + assertEquals(1, SampleNotThreadSafeCases.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..9c42dee00d26 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 @@ -35,6 +35,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 +57,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(); 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..fcce9ccec574 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 @@ -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,7 +1591,7 @@ 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; @@ -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(); From 76b0857783c822b46b003459e3887348da165f35 Mon Sep 17 00:00:00 2001 From: Reuven Lax Date: Tue, 29 Sep 2026 13:33:49 -0700 Subject: [PATCH 2/5] fix or disable tests that cannot be parallelizes --- .../org/apache/beam/runners/dataflow/DataflowRunnerTest.java | 4 ++-- .../core/src/test/java/org/apache/beam/sdk/PipelineTest.java | 3 +++ .../test/java/org/apache/beam/sdk/metrics/MetricsTest.java | 1 + .../java/org/apache/beam/sdk/transforms/CombineFnsTest.java | 5 +++-- .../test/java/org/apache/beam/sdk/transforms/ParDoTest.java | 4 ++-- 5 files changed, 11 insertions(+), 6 deletions(-) diff --git a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java index 6ee47109c082..549e4f782a39 100644 --- a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java +++ b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java @@ -142,7 +142,6 @@ 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.ExpectedLogs; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; @@ -206,6 +205,7 @@ import org.junit.rules.ExpectedException; import org.junit.rules.TemporaryFolder; import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; import org.mockito.ArgumentCaptor; import org.mockito.Mockito; @@ -214,7 +214,7 @@ * *

Implements {@link Serializable} because it is caught in closures. */ -@RunWith(BeamParallelJunit4Runner.class) +@RunWith(JUnit4.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/PipelineTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/PipelineTest.java index 6ddd854126cc..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 @@ -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/metrics/MetricsTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/MetricsTest.java index 539905b321a0..b49ca6c36dea 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 @@ -274,6 +274,7 @@ public void testCounterToCell() { } @Test + @BeamParallelJunit4Runner.SerialTest public void testMetricsFlag() { Metrics.resetDefaultPipelineOptions(); assertFalse(Metrics.MetricsFlag.counterDisabled()); 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 9c42dee00d26..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; @@ -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/ParDoTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java index fcce9ccec574..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; @@ -1597,7 +1597,7 @@ private abstract static class BundleFinalizingDoFn extends DoFn 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) From 52cc4c2af976047713471a98cb230a3d9266000a Mon Sep 17 00:00:00 2001 From: Reuven Lax Date: Tue, 29 Sep 2026 13:40:14 -0700 Subject: [PATCH 3/5] simplify --- .../sdk/testing/BeamParallelJunit4Runner.java | 52 ++----------------- .../testing/BeamParallelJunit4RunnerTest.java | 11 ++-- 2 files changed, 9 insertions(+), 54 deletions(-) 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 index 805383cef2f2..6527fb487b65 100644 --- 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 @@ -17,18 +17,13 @@ */ package org.apache.beam.sdk.testing; -import java.io.IOException; -import java.io.InputStream; -import java.lang.annotation.Annotation; import java.lang.annotation.ElementType; import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; -import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Collections; import java.util.List; -import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; @@ -36,8 +31,6 @@ import java.util.concurrent.Future; import java.util.concurrent.atomic.AtomicInteger; import org.apache.beam.sdk.annotations.Internal; -import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableSet; -import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.io.ByteStreams; import org.checkerframework.checker.nullness.qual.Nullable; import org.junit.experimental.categories.Category; import org.junit.runner.notification.RunNotifier; @@ -58,10 +51,8 @@ * all other {@code @Test} methods (defaults to {@code 1}, i.e., sequential execution). * * - *

Classes or methods annotated with {@link SerialTest}, {@code @NotThreadSafe} (such as {@code - * javax.annotation.concurrent.NotThreadSafe} or {@code net.jcip.annotations.NotThreadSafe}), - * {@code @Serial}, or {@code @Isolated} are always executed sequentially on the calling thread - * after draining any in-flight parallel test methods in the class. + *

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 { @@ -78,9 +69,6 @@ public final class BeamParallelJunit4Runner extends BlockJUnit4ClassRunner { "beam.validatesRunner.parallelThreads"; public static final String DEFAULT_TEST_THREADS_PROPERTY = "beam.test.parallelThreads"; - private static final Set SERIAL_ANNOTATION_SIMPLE_NAMES = - ImmutableSet.of("NotThreadSafe", "SerialTest", "Serial", "Isolated"); - private static final ConcurrentHashMap EXECUTORS = new ConcurrentHashMap<>(); @@ -203,7 +191,7 @@ private static boolean hasValidatesRunnerCategory(@Nullable Category category) { } private static boolean isMethodMarkedSerial(FrameworkMethod method) { - return hasSerialAnnotation(method.getAnnotations()); + return method.getAnnotation(SerialTest.class) != null; } private static boolean isClassMarkedSerial(@Nullable Class clazz) { @@ -217,40 +205,8 @@ private static boolean computeClassSerial(@Nullable Class clazz) { if (clazz == null || clazz == Object.class) { return false; } - return hasSerialAnnotation(clazz.getAnnotations()) - || hasClassFileNotThreadSafeAnnotation(clazz) + return clazz.isAnnotationPresent(SerialTest.class) || computeClassSerial(clazz.getSuperclass()) || computeClassSerial(clazz.getEnclosingClass()); } - - private static boolean hasSerialAnnotation(Annotation[] annotations) { - for (Annotation annotation : annotations) { - if (SERIAL_ANNOTATION_SIMPLE_NAMES.contains(annotation.annotationType().getSimpleName())) { - return true; - } - } - return false; - } - - /** - * Inspects the {@code .class} bytecode for {@code RetentionPolicy.CLASS} annotations such as - * {@code javax.annotation.concurrent.NotThreadSafe} or {@code net.jcip.annotations.NotThreadSafe} - * that are not retained for runtime reflection via {@link Class#getAnnotations()}. - */ - private static boolean hasClassFileNotThreadSafeAnnotation(Class clazz) { - String resourceName = clazz.getName().replace('.', '/') + ".class"; - ClassLoader classLoader = clazz.getClassLoader(); - try (InputStream in = - classLoader != null - ? classLoader.getResourceAsStream(resourceName) - : ClassLoader.getSystemResourceAsStream(resourceName)) { - if (in == null) { - return false; - } - String raw = new String(ByteStreams.toByteArray(in), StandardCharsets.ISO_8859_1); - return raw.contains("/NotThreadSafe;"); - } catch (IOException e) { - return false; - } - } } 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 index f5912e2de14c..1fd2b168bd06 100644 --- 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 @@ -27,7 +27,6 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; -import javax.annotation.concurrent.NotThreadSafe; import org.apache.beam.sdk.testing.BeamParallelJunit4Runner.SerialTest; import org.junit.Rule; import org.junit.Test; @@ -93,9 +92,9 @@ public void testUnit2() throws InterruptedException { } } - @NotThreadSafe + @SerialTest @RunWith(BeamParallelJunit4Runner.class) - public static class SampleNotThreadSafeCases { + public static class SampleSerialClassCases { static final AtomicInteger ACTIVE = new AtomicInteger(0); static final AtomicInteger MAX_ACTIVE = new AtomicInteger(0); @@ -173,15 +172,15 @@ public void testSeparateParallelismForValidatesRunnerVsUnitTests() { } @Test - public void testNotThreadSafeClassRunsSerially() { + public void testSerialTestClassRunsSerially() { System.setProperty(BeamParallelJunit4Runner.VALIDATES_RUNNER_THREADS_PROPERTY, "4"); IN_TEST_HARNESS.set(true); try { - Result result = JUnitCore.runClasses(SampleNotThreadSafeCases.class); + Result result = JUnitCore.runClasses(SampleSerialClassCases.class); assertEquals(0, result.getFailureCount()); assertEquals(2, result.getRunCount()); - assertEquals(1, SampleNotThreadSafeCases.MAX_ACTIVE.get()); + assertEquals(1, SampleSerialClassCases.MAX_ACTIVE.get()); } finally { IN_TEST_HARNESS.set(false); } From 89dd9297baab7b9d78429be8874599a99f88aa98 Mon Sep 17 00:00:00 2001 From: Reuven Lax Date: Tue, 29 Sep 2026 13:58:07 -0700 Subject: [PATCH 4/5] remove unrelated changes --- .../sdk/testing/BeamParallelJunit4Runner.java | 6 +- .../apache/beam/sdk/metrics/MetricsTest.java | 157 ++++++++---------- 2 files changed, 74 insertions(+), 89 deletions(-) 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 index 6527fb487b65..7fc8552c50b0 100644 --- 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 @@ -154,6 +154,9 @@ protected void runChild(final FrameworkMethod method, final RunNotifier 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; @@ -165,9 +168,6 @@ protected void runChild(final FrameworkMethod method, final RunNotifier notifier int defaultThreads = Integer.getInteger(DEFAULT_TEST_THREADS_PROPERTY, 1); if (isValidatesRunnerMethod(method)) { int vrThreads = Integer.getInteger(VALIDATES_RUNNER_THREADS_PROPERTY, defaultThreads); - if (vrThreads == defaultThreads) { - return getOrCreateExecutor("beam-test-worker", defaultThreads); - } return getOrCreateExecutor("beam-vr-worker", vrThreads); } return getOrCreateExecutor("beam-test-worker", defaultThreads); 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 b49ca6c36dea..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 @@ -34,7 +34,6 @@ import java.io.Serializable; import java.util.List; import java.util.NoSuchElementException; -import java.util.concurrent.ConcurrentHashMap; import org.apache.beam.sdk.PipelineResult; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.coders.VarIntCoder; @@ -92,9 +91,6 @@ private static MetricQueryResults queryTestMetrics(PipelineResult result) { /** Shared test helpers and setup/teardown. */ public abstract static class SharedTestBase implements Serializable { - private static final ConcurrentHashMap CACHED_METRIC_PIPELINE_RESULTS = - new ConcurrentHashMap<>(); - @Rule public final transient ExpectedException thrown = ExpectedException.none(); @Rule public final transient TestPipeline pipeline = TestPipeline.create(); @@ -105,88 +101,77 @@ public void tearDown() { } protected PipelineResult runPipelineWithMetrics() { - String cacheKey = - pipeline.getOptions().getRunner().getName() - + ":" - + System.getProperty(TestPipeline.PROPERTY_BEAM_TEST_PIPELINE_OPTIONS, ""); - synchronized (CACHED_METRIC_PIPELINE_RESULTS) { - PipelineResult cachedResult = CACHED_METRIC_PIPELINE_RESULTS.get(cacheKey); - if (cachedResult != null) { - return cachedResult; - } - final Counter count = Metrics.counter(MetricsTest.class, "count"); - StringSet sideinputs = Metrics.stringSet(MetricsTest.class, "sideinputs"); - final TupleTag output1 = new TupleTag() {}; - final TupleTag output2 = new TupleTag() {}; - pipeline - .apply(Create.of(5, 8, 13)) - .apply( - "MyStep1", - ParDo.of( - new DoFn() { - Distribution bundleDist = Metrics.distribution(MetricsTest.class, "bundle"); - - @StartBundle - public void startBundle() { - bundleDist.update(10L); - } - - @SuppressWarnings("unused") - @ProcessElement - public void processElement(ProcessContext c) { - Distribution values = Metrics.distribution(MetricsTest.class, "input"); - StringSet sources = Metrics.stringSet(MetricsTest.class, "sources"); - BoundedTrie boundedTrieSources = - Metrics.boundedTrie(MetricsTest.class, "boundedTrieSources"); - count.inc(); - values.update(c.element()); - - c.output(c.element()); - c.output(c.element()); - sources.add("gcs"); - sources.add("gcs"); // repeated should appear once - sources.add("gcs", "gcs"); // repeated should appear once - sideinputs.add("bigtable", "spanner"); - boundedTrieSources.add(ImmutableList.of("ab_source", "cd_source")); - boundedTrieSources.add(ImmutableList.of("ef_source")); - } - - @DoFn.FinishBundle - public void finishBundle() { - bundleDist.update(40L); - } - })) - .apply( - "MyStep2", - ParDo.of( - new DoFn() { - @SuppressWarnings("unused") - @ProcessElement - public void processElement(ProcessContext c) { - Distribution values = Metrics.distribution(MetricsTest.class, "input"); - Gauge gauge = Metrics.gauge(MetricsTest.class, "my-gauge"); - StringSet sinks = Metrics.stringSet(MetricsTest.class, "sinks"); - BoundedTrie boundedTrieSinks = - Metrics.boundedTrie(MetricsTest.class, "boundedTrieSinks"); - Integer element = c.element(); - count.inc(); - values.update(element); - gauge.set(12L); - c.output(element); - sinks.add("bq", "kafka", "kafka"); // repeated should appear once - sideinputs.add("bigtable", "sql"); - boundedTrieSinks.add(ImmutableList.of("ab_sink", "cd_sink")); - boundedTrieSinks.add(ImmutableList.of("ef_sink")); - c.output(output2, element); - } - }) - .withOutputTags(output1, TupleTagList.of(output2))); - PipelineResult result = pipeline.run(); - - result.waitUntilFinish(); - CACHED_METRIC_PIPELINE_RESULTS.put(cacheKey, result); - return result; - } + final Counter count = Metrics.counter(MetricsTest.class, "count"); + StringSet sideinputs = Metrics.stringSet(MetricsTest.class, "sideinputs"); + final TupleTag output1 = new TupleTag() {}; + final TupleTag output2 = new TupleTag() {}; + pipeline + .apply(Create.of(5, 8, 13)) + .apply( + "MyStep1", + ParDo.of( + new DoFn() { + Distribution bundleDist = Metrics.distribution(MetricsTest.class, "bundle"); + + @StartBundle + public void startBundle() { + bundleDist.update(10L); + } + + @SuppressWarnings("unused") + @ProcessElement + public void processElement(ProcessContext c) { + Distribution values = Metrics.distribution(MetricsTest.class, "input"); + StringSet sources = Metrics.stringSet(MetricsTest.class, "sources"); + BoundedTrie boundedTrieSources = + Metrics.boundedTrie(MetricsTest.class, "boundedTrieSources"); + count.inc(); + values.update(c.element()); + + c.output(c.element()); + c.output(c.element()); + sources.add("gcs"); + sources.add("gcs"); // repeated should appear once + sources.add("gcs", "gcs"); // repeated should appear once + sideinputs.add("bigtable", "spanner"); + boundedTrieSources.add(ImmutableList.of("ab_source", "cd_source")); + boundedTrieSources.add(ImmutableList.of("ef_source")); + } + + @DoFn.FinishBundle + public void finishBundle() { + bundleDist.update(40L); + } + })) + .apply( + "MyStep2", + ParDo.of( + new DoFn() { + @SuppressWarnings("unused") + @ProcessElement + public void processElement(ProcessContext c) { + Distribution values = Metrics.distribution(MetricsTest.class, "input"); + Gauge gauge = Metrics.gauge(MetricsTest.class, "my-gauge"); + StringSet sinks = Metrics.stringSet(MetricsTest.class, "sinks"); + BoundedTrie boundedTrieSinks = + Metrics.boundedTrie(MetricsTest.class, "boundedTrieSinks"); + Integer element = c.element(); + count.inc(); + values.update(element); + gauge.set(12L); + c.output(element); + sinks.add("bq", "kafka", "kafka"); // repeated should appear once + sideinputs.add("bigtable", "sql"); + boundedTrieSinks.add(ImmutableList.of("ab_sink", "cd_sink")); + boundedTrieSinks.add(ImmutableList.of("ef_sink")); + c.output(output2, element); + } + }) + .withOutputTags(output1, TupleTagList.of(output2))); + PipelineResult result = pipeline.run(); + + result.waitUntilFinish(); + return result; } } From 6a3c4daf6e4830f690e8f636b71bd9648997ba1d Mon Sep 17 00:00:00 2001 From: Reuven Lax Date: Tue, 29 Sep 2026 16:53:44 -0700 Subject: [PATCH 5/5] use CompleteableFuture --- .../sdk/testing/BeamParallelJunit4Runner.java | 43 +++++-------------- 1 file changed, 10 insertions(+), 33 deletions(-) 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 index 7fc8552c50b0..bcaead005add 100644 --- 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 @@ -21,14 +21,10 @@ import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; -import java.util.ArrayList; -import java.util.Collections; -import java.util.List; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; -import java.util.concurrent.Future; import java.util.concurrent.atomic.AtomicInteger; import org.apache.beam.sdk.annotations.Internal; import org.checkerframework.checker.nullness.qual.Nullable; @@ -75,7 +71,7 @@ public final class BeamParallelJunit4Runner extends BlockJUnit4ClassRunner { private static final ConcurrentHashMap, Boolean> CLASS_SERIAL_CACHE = new ConcurrentHashMap<>(); - private final List> pendingFutures = Collections.synchronizedList(new ArrayList<>()); + private CompletableFuture pendingFutures = CompletableFuture.allOf(); private static @Nullable ExecutorService getOrCreateExecutor(String poolKey, int threads) { if (threads <= 1) { @@ -113,32 +109,10 @@ public void finished() { } private void awaitPendingFutures() { - Throwable firstError = null; - List> snapshot; - synchronized (pendingFutures) { - snapshot = new ArrayList<>(pendingFutures); - pendingFutures.clear(); - } - for (Future future : snapshot) { - try { - future.get(); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - if (firstError == null) { - firstError = e; - } - } catch (ExecutionException e) { - if (firstError == null) { - firstError = e.getCause() != null ? e.getCause() : e; - } - } - } - if (firstError instanceof RuntimeException) { - throw (RuntimeException) firstError; - } else if (firstError instanceof Error) { - throw (Error) firstError; - } else if (firstError != null) { - throw new RuntimeException(firstError); + try { + pendingFutures.join(); + } finally { + pendingFutures = CompletableFuture.allOf(); } } @@ -161,7 +135,10 @@ protected void runChild(final FrameworkMethod method, final RunNotifier notifier super.runChild(method, notifier); return; } - pendingFutures.add(executor.submit(() -> super.runChild(method, notifier))); + pendingFutures = + CompletableFuture.allOf( + pendingFutures, + CompletableFuture.runAsync(() -> super.runChild(method, notifier), executor)); } private @Nullable ExecutorService selectExecutor(FrameworkMethod method) {