[GitHub] [beam] chamikaramj commented on a change in pull request #15249: [BEAM-12690] Fix GroupIntoBatches watermark maintenance

[GitBox <gi...@apache.org>]

chamikaramj commented on a change in pull request #15249:
URL: https://github.com/apache/beam/pull/15249#discussion_r683803943



##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/state/Timer.java
##########
@@ -100,9 +100,9 @@
   Timer withOutputTimestamp(Instant outputTime);
 
   /**
-   * Returns the current relative time used by {@link #setRelative()} and {@link #offset}. This can
-   * be used by a client that self-manages relative timers (e.g. one that stores the current timer
-   * time in a state variable.
+   * * Returns the current relative time used by {@link #setRelative()} and {@link #offset}. This
+   * can be used by a client that self-manages relative timers (e.g. one that stores the current

Review comment:
       Seems like an unintended formatting change ?

##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupIntoBatches.java
##########
@@ -398,11 +424,31 @@ public void processElement(
         storedBatchSizeBytes.readLater();
       }
 
-      long num = storedBatchSize.read();
-      if (maxBufferingDuration.isLongerThan(Duration.ZERO) && num == 1) {
-        // This is the first element in batch. Start counting buffering time if a limit was set.
-        bufferingTimer.offset(maxBufferingDuration).setRelative();
+      long num;
+      if (maxBufferingDuration.isLongerThan(Duration.ZERO)) {
+        minBufferedTs.readLater();
+        num = storedBatchSize.read();
+
+        long oldOutputTs =
+            MoreObjects.firstNonNull(
+                minBufferedTs.read(), BoundedWindow.TIMESTAMP_MAX_VALUE.getMillis());
+        minBufferedTs.add(elementTs.getMillis());
+        // If this is the first element in the batch or if the timer's output timestamp needs
+        // modifying, then set a
+        // timer.
+        if (num == 1 || minBufferedTs.read() != oldOutputTs) {
+          long targetTs =
+              MoreObjects.firstNonNull(
+                  timerTs.read(),
+                  bufferingTimer.getCurrentRelativeTime().getMillis()

Review comment:
       I might be misunderstanding something but why can't we just set this to "current time" + "maxBufferingDuration" ?

##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupIntoBatches.java
##########
@@ -340,6 +342,16 @@ public long getElementByteSize() {
     @StateId(NUM_BYTES_IN_BATCH_ID)
     private final StateSpec<CombiningState<Long, long[], Long>> batchSizeBytesSpec;
 
+    private static final String TIMER_TIMESTAMP = "timerTs";
+
+    @StateId(TIMER_TIMESTAMP)
+    private final StateSpec<ValueState<Long>> timerTsSpec;

Review comment:
       Probably we should add comments clarifying what each of these state Ids are for.




-- 
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: github-unsubscribe@beam.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org