[acrites]

Watermark Hold for Out-of-Order Elements:

• Previously, the timer watermark hold was only set when buffering the first element (storedBatchSize.read() == 1). If subsequent elements arrived with timestamps earlier than the first element (out-of-order within the window), the watermark hold remained pinned to the first element's timestamp, risking premature watermark advancement.
• Now, GroupIntoBatches compares currentMinBufferedTs against oldMinBufferedTs and updates the timer's output timestamp whenever an earlier timestamp arrives (num == 1 || currentMinBufferedTs < oldMinBufferedTs).

Tests:

• Added testOutOfOrderTimestampsInBatch verifying out of order elements don't get dropped as late data.

[github-actions]

📋 Review Summary

This PR fixes several important bugs in GroupIntoBatches, including updating timer watermark holds for out-of-order elements, persisting timerTs to prevent processing-time timer starvation, and preventing timestamp overflow on window expiration calculations. It also adds display data support, improves precondition checks, and fixes Javadoc formatting.

🔍 General Feedback

• The state management and watermark hold logic in GroupIntoBatchesDoFn are clean and properly handle out-of-order elements without resetting the processing-time deadline.
• testOutOfOrderTimestampsInBatch could be strengthened to directly verify the updated timer watermark hold (Right now, a batch size of 2 flushes immediately inside processElement, which clears the timer before the watermark advances).

---

💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.

[github-actions]

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

[acrites]

R: @arunpandianp

I made the changes more focused. There were some PreCommit failures, but I don't see anything related to GroupIntoBatches in the failing tests.

[github-actions]

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment assign set of reviewers

[arunpandianp]

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.

[acrites]

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.

[arunpandianp]

Thanks for fixing this!

[arunpandianp]

nit: i like currentMinBufferedTs != oldOutputTs, set a timer when MinBufferedTs is modified is easier to read than when currentMinBufferedTs < oldOutputTs. they both are true when we need a new timer.