lukecwik commented on a change in pull request #12488:
URL: https://github.com/apache/beam/pull/12488#discussion_r467358516
##########
File path:
sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
##########
@@ -1499,6 +1504,146 @@ public void populateDisplayData(Builder builder) {
}
}
+ @RunWith(JUnit4.class)
+ public static class BundleFinalizationTests extends SharedTestBase
implements Serializable {
+ private abstract static class BundleFinalizingDoFn extends DoFn<KV<String,
Long>, String> {
+ private static final long MAX_ATTEMPTS = 3000;
+ // We use the UUID to uniquely identify this DoFn in case this test is
run with
+ // other tests in the same JVM.
+ private static final Map<UUID, AtomicBoolean> WAS_FINALIZED = new
HashMap();
+ private final UUID uuid = UUID.randomUUID();
+
+ public void testFinalization(BundleFinalizer bundleFinalizer,
OutputReceiver<String> output)
+ throws Exception {
+ if (WAS_FINALIZED.computeIfAbsent(uuid, (unused) -> new
AtomicBoolean()).get()) {
+ output.output("bundle was finalized");
+ return;
+ }
+ bundleFinalizer.afterBundleCommit(
+ Instant.now().plus(Duration.standardSeconds(MAX_ATTEMPTS)),
+ () -> WAS_FINALIZED.computeIfAbsent(uuid, (unused) -> new
AtomicBoolean()).set(true));
+ // We sleep here to give time for the runner to perform any prior
callbacks.
+ sleep(100L);
Review comment:
MAX_ATTEMPTS * timeout >= 300 seconds. I don't believe it should flake
but if it does I'll have to figure out a different way to design this test.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
