github-actions[bot] commented on code in PR #40754:
URL: https://github.com/apache/beam/pull/40754#discussion_r4362894738
##########
sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupIntoBatchesTest.java:
##########
@@ -805,4 +808,66 @@ private static <K> boolean checkBatchByteSizes(
}
return true;
}
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testMaxBatchSizeThrows() {
+ GroupIntoBatches.ofSize(Long.MAX_VALUE);
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testMaxBatchSizeBytesThrows() {
+ GroupIntoBatches.ofByteSize(Long.MAX_VALUE);
+ }
+
+ @Test
+ public void testDisplayData() {
+ GroupIntoBatches<String, String> gib =
+ GroupIntoBatches.<String, String>ofSize(100)
+ .withByteSize(1000)
+ .withMaxBufferingDuration(Duration.standardSeconds(5));
+ DisplayData displayData = DisplayData.from(gib);
+ assertThat(displayData, hasDisplayItem("batchSize", 100L));
+ assertThat(displayData, hasDisplayItem("batchSizeBytes", 1000L));
+ assertThat(displayData, hasDisplayItem("maxBufferingDuration",
Duration.standardSeconds(5)));
+ }
+
+ @Test
+ @Category({
+ ValidatesRunner.class,
+ NeedsRunner.class,
+ UsesTimersInParDo.class,
+ UsesStatefulParDo.class,
+ UsesTestStream.class,
+ UsesOnWindowExpiration.class
+ })
+ public void testOutOfOrderTimestampsInBatch() {
+ Instant baseTs = new Instant(0L);
+ TestStream<KV<String, String>> stream =
+ TestStream.create(KvCoder.of(StringUtf8Coder.of(),
StringUtf8Coder.of()))
+ .advanceWatermarkTo(baseTs)
+ .addElements(
+ TimestampedValue.of(
+ KV.of("key", "val1"),
baseTs.plus(Duration.standardSeconds(10))))
+ .addElements(
+ TimestampedValue.of(KV.of("key", "val2"),
baseTs.plus(Duration.standardSeconds(5))))
+ .advanceWatermarkToInfinity();
+
+ PCollection<Long> outputTimestamps =
+ pipeline
+ .apply(stream)
+ .apply(GroupIntoBatches.ofSize(2))
Review Comment:
**🟢 Low**: With `ofSize(2)` and 2 input elements, the batch is flushed
immediately in `processElement` when the second element arrives (which uses
`minBufferedTs` directly and clears `holdTimer`), so the timer's watermark hold
is never actually exercised on window expiration or timer firing. Setting the
batch size to `3` (so the 2 elements remain buffered until the watermark
advances to infinity and triggers `@OnWindowExpiration`) would directly
exercise the updated timer watermark hold logic.
```suggestion
.apply(GroupIntoBatches.ofSize(3))
```
--
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]
