acrites commented on code in PR #40754:
URL: https://github.com/apache/beam/pull/40754#discussion_r4383537829
##########
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 elements were getting output with the wrong timestamps. Let me modify
the test so that there is something like a GBK afterwards and we can directly
verify that the elements get dropped due to an incorrect watermark hold. In the
new example, the batch gets emitted with timestamp 2s, but the watermark hold
was set to 10s.
--
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]
