From 184f45df7d58dd65d102911cf7782cce5665c316 Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Tue, 30 Jun 2026 13:30:14 +0000 Subject: [PATCH 1/3] fix flakey fileiotest --- .../org/apache/beam/sdk/io/FileIOTest.java | 53 +++++++++++++++++-- 1 file changed, 49 insertions(+), 4 deletions(-) diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/FileIOTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/FileIOTest.java index dffc4943bfab..afbc6200b015 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/FileIOTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/FileIOTest.java @@ -271,9 +271,46 @@ public void processElement(ProcessContext context, @StateId("count") ValueState< private final String watchPathStr; } + private static class AfterNumberOfNewOutputs + implements Watch.Growth.TerminationCondition { + private final int numOutputs; + + public AfterNumberOfNewOutputs(int numOutputs) { + this.numOutputs = numOutputs; + } + + @Override + public org.apache.beam.sdk.coders.Coder getStateCoder() { + return VarIntCoder.of(); + } + + @Override + public Integer forNewInput(org.joda.time.Instant now, String input) { + return 0; + } + + @Override + public Integer onSeenNewOutput(org.joda.time.Instant now, Integer state) { + return state + 1; + } + + @Override + public boolean canStopPolling(org.joda.time.Instant now, Integer state) { + return state >= numOutputs; + } + + @Override + public String toString(Integer state) { + return "AfterNumberOfNewOutputs(" + state + "/" + numOutputs + ")"; + } + } + @Test @Category({NeedsRunner.class, UsesUnboundedSplittableParDo.class}) public void testMatchWatchForNewFiles() throws IOException, InterruptedException { + final int numFiles = 3; + final int numUpdatedFiles = 6; // first x3, second x2, third x1 + // Write some files to a "source" directory. final Path sourcePath = tmpFolder.getRoot().toPath().resolve("source"); sourcePath.toFile().mkdir(); @@ -291,7 +328,9 @@ public void testMatchWatchForNewFiles() throws IOException, InterruptedException .filepattern(watchPath.resolve("*").toString()) .continuously( Duration.millis(100), - Watch.Growth.afterTimeSinceNewOutput(Duration.standardSeconds(1)))); + Watch.Growth.eitherOf( + new AfterNumberOfNewOutputs(numFiles), + Watch.Growth.afterTimeSinceNewOutput(Duration.standardSeconds(10))))); PCollection matchAllMetadata = p.apply("create for matchAll new files", Create.of(watchPath.resolve("*").toString())) .apply( @@ -299,7 +338,9 @@ public void testMatchWatchForNewFiles() throws IOException, InterruptedException FileIO.matchAll() .continuously( Duration.millis(100), - Watch.Growth.afterTimeSinceNewOutput(Duration.standardSeconds(1)))); + Watch.Growth.eitherOf( + new AfterNumberOfNewOutputs(numFiles), + Watch.Growth.afterTimeSinceNewOutput(Duration.standardSeconds(10))))); PCollection matchUpdatedMetadata = p.apply( "match updated", @@ -307,7 +348,9 @@ public void testMatchWatchForNewFiles() throws IOException, InterruptedException .filepattern(watchPath.resolve("first").toString()) .continuously( Duration.millis(100), - Watch.Growth.afterTimeSinceNewOutput(Duration.standardSeconds(1)), + Watch.Growth.eitherOf( + new AfterNumberOfNewOutputs(numFiles), + Watch.Growth.afterTimeSinceNewOutput(Duration.standardSeconds(10))), true)); PCollection matchAllUpdatedMetadata = p.apply("create for matchAll updated files", Create.of(watchPath.resolve("*").toString())) @@ -316,7 +359,9 @@ public void testMatchWatchForNewFiles() throws IOException, InterruptedException FileIO.matchAll() .continuously( Duration.millis(100), - Watch.Growth.afterTimeSinceNewOutput(Duration.standardSeconds(1)), + Watch.Growth.eitherOf( + new AfterNumberOfNewOutputs(numUpdatedFiles), + Watch.Growth.afterTimeSinceNewOutput(Duration.standardSeconds(10))), true)); // write one file at the beginning. This will trigger the first output for matchAll From fc4cf122a2689799e418a83ef4b51b45efeb21df Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Tue, 30 Jun 2026 16:27:21 +0000 Subject: [PATCH 2/3] fix gemini --- .../core/src/test/java/org/apache/beam/sdk/io/FileIOTest.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/FileIOTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/FileIOTest.java index afbc6200b015..a816d4c2615b 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/FileIOTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/FileIOTest.java @@ -309,6 +309,7 @@ public String toString(Integer state) { @Category({NeedsRunner.class, UsesUnboundedSplittableParDo.class}) public void testMatchWatchForNewFiles() throws IOException, InterruptedException { final int numFiles = 3; + final int numFirstFiles = 3; final int numUpdatedFiles = 6; // first x3, second x2, third x1 // Write some files to a "source" directory. @@ -349,7 +350,7 @@ public void testMatchWatchForNewFiles() throws IOException, InterruptedException .continuously( Duration.millis(100), Watch.Growth.eitherOf( - new AfterNumberOfNewOutputs(numFiles), + new AfterNumberOfNewOutputs(numFirstFiles), Watch.Growth.afterTimeSinceNewOutput(Duration.standardSeconds(10))), true)); PCollection matchAllUpdatedMetadata = From 444a8ff97582adf2a3da4ca297ede7ebf1df8b6f Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Tue, 30 Jun 2026 16:53:08 +0000 Subject: [PATCH 3/3] more gemini fixes --- .../core/src/test/java/org/apache/beam/sdk/io/FileIOTest.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/FileIOTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/FileIOTest.java index a816d4c2615b..2cffce762135 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/FileIOTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/FileIOTest.java @@ -291,12 +291,12 @@ public Integer forNewInput(org.joda.time.Instant now, String input) { @Override public Integer onSeenNewOutput(org.joda.time.Instant now, Integer state) { - return state + 1; + return (state == null ? 0 : state) + 1; } @Override public boolean canStopPolling(org.joda.time.Instant now, Integer state) { - return state >= numOutputs; + return (state == null ? 0 : state) >= numOutputs; } @Override