arunpandianp commented on code in PR #40754:
URL: https://github.com/apache/beam/pull/40754#discussion_r4383260623
##########
sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupIntoBatchesTest.java:
##########
@@ -805,4 +805,49 @@ private static <K> boolean checkBatchByteSizes(
}
return true;
}
+
+ /**
+ * Tests that when elements arrive out of order, the watermark hold is
updated to the minimum
+ * timestamp. With ofSize(3) and 2 elements, elements remain buffered until
window expiration,
+ * exercising the updated timer watermark hold rather than flushing
immediately in processElement.
+ */
+ @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(3))
+ .apply(
+ ParDo.of(
+ new DoFn<KV<String, Iterable<String>>, Long>() {
+ @ProcessElement
+ public void processElement(
+ @Timestamp Instant ts, OutputReceiver<Long>
receiver) {
+ receiver.output(ts.getMillis());
Review Comment:
the fix looks good to me. Is the test failing without the fix?
It seems like receiver.output will get fused with the GroupIntoBatches timer
and i'm not sure if the elements will be dropped as late.
--
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]
