gemini-code-assist[bot] commented on code in PR #38320:
URL: https://github.com/apache/beam/pull/38320#discussion_r3171085429
##########
sdks/java/core/src/test/java/org/apache/beam/sdk/io/FileIOTest.java:
##########
@@ -376,6 +349,35 @@ public void testMatchWatchForNewFiles() throws
IOException, InterruptedException
p.run();
}
+ @Test
+ public void testMatchWatchForNewFiles_UnboundedPCollection() {
+ // Additional scenarios for testMatchWatchForNewFiles. Only construct the
pipeline and check
+ // output pcoll
+ final Path watchPath = tmpFolder.getRoot().toPath().resolve("watch");
+
+ PCollection<MatchResult.Metadata> matchAllMetadata =
+ p.apply("create for matchAll new files", Create.of(watchPath + "/*"))
+ .apply(
+ "match filename through matchAll",
+ FileIO.matchAll()
+ .continuously(
+ Duration.millis(100),
+
Watch.Growth.afterTimeSinceNewOutput(Duration.standardSeconds(1))));
+
+ PCollection<MatchResult.Metadata> matchUpdatedMetadata =
+ p.apply(
+ "match updated",
+ FileIO.match()
+ .filepattern(watchPath.resolve("first").toString())
+ .continuously(
+ Duration.millis(100),
+
Watch.Growth.afterTimeSinceNewOutput(Duration.standardSeconds(1)),
+ true));
+
+ assertEquals(PCollection.IsBounded.UNBOUNDED,
matchAllMetadata.isBounded());
+ assertEquals(PCollection.IsBounded.UNBOUNDED,
matchUpdatedMetadata.isBounded());
+ }
Review Comment:
This new test method has two significant issues:
1. **Compilation Error**: At line 359, `watchPath + "/*"` will not compile
because `watchPath` is a `java.nio.file.Path` object and Java does not support
the `+` operator for `Path` and `String`. It should use
`watchPath.resolve("*").toString()` to be correct and consistent with the rest
of the file.
2. **Test Failure (Abandoned Nodes)**: The test uses the `TestPipeline` rule
`p` but does not call `p.run()`. By default, `TestPipeline` throws an exception
if nodes are added but the pipeline is not executed. Since this test only
checks the boundedness of the PCollections, you should disable the abandoned
node enforcement using `p.enableAbandonedNodeEnforcement(false)`.
Here is the corrected method:
```java
public void testMatchWatchForNewFiles_UnboundedPCollection() {
// Additional scenarios for testMatchWatchForNewFiles. Only construct
the pipeline and check
// output pcoll
p.enableAbandonedNodeEnforcement(false);
final Path watchPath = tmpFolder.getRoot().toPath().resolve("watch");
PCollection<MatchResult.Metadata> matchAllMetadata =
p.apply("create for matchAll new files",
Create.of(watchPath.resolve("*").toString()))
.apply(
"match filename through matchAll",
FileIO.matchAll()
.continuously(
Duration.millis(100),
Watch.Growth.afterTimeSinceNewOutput(Duration.standardSeconds(1))));
PCollection<MatchResult.Metadata> matchUpdatedMetadata =
p.apply(
"match updated",
FileIO.match()
.filepattern(watchPath.resolve("first").toString())
.continuously(
Duration.millis(100),
Watch.Growth.afterTimeSinceNewOutput(Duration.standardSeconds(1)),
true));
assertEquals(PCollection.IsBounded.UNBOUNDED,
matchAllMetadata.isBounded());
assertEquals(PCollection.IsBounded.UNBOUNDED,
matchUpdatedMetadata.isBounded());
}
```
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
