[acrites]

1. 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).
2. Persistent timerTs State to Prevent Timer Starvation:
   • When updating the timer's watermark hold for out-of-order elements, calculating a new firing time via getCurrentRelativeTime() + maxBufferingDuration on every update would push the timer deadline further out, causing timer starvation.
   • Uses ValueState<Long> timerTs, which had a bug where it was never actually written, to store the target firing timestamp calculated on the first element. When updating the watermark hold, the target firing time is preserved.
3. Window Overflow Protection:
   • Guarded window.maxTimestamp().plus(allowedLateness) when setting holdTimer to prevent Instant arithmetic overflow when close to BoundedWindow.TIMESTAMP_MAX_VALUE.
4. Display Data:
   • Implemented populateDisplayData on GroupIntoBatches, GroupIntoBatches.WithShardedKey, and GroupIntoBatchesDoFn to expose batchSize, batchSizeBytes, and maxBufferingDuration in runner UIs (such as Dataflow).
5. Validation & Javadoc Cleanup:
   • Replaced Preconditions.checkState with checkArgument with informative error messages for batchSize < Long.MAX_VALUE and batchSizeBytes < Long.MAX_VALUE.
   • Fixed malformed syntax and stray braces in class-level Javadoc code snippets.
6. Tests:
   • Added testOutOfOrderTimestampsInBatch verifying watermark hold output timestamp matches the minimum timestamp of out-of-order elements.
   • Added tests for checkArgument bounds and populateDisplayData.

[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