arunpandianp commented on code in PR #40754:
URL: https://github.com/apache/beam/pull/40754#discussion_r4383657686
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupIntoBatches.java:
##########
@@ -527,20 +524,22 @@ public void processElement(
}
storedBatchSizeBytes.add(elementWeight);
}
+
+ long oldOutputTs =
+ MoreObjects.firstNonNull(
+ minBufferedTs.read(),
BoundedWindow.TIMESTAMP_MAX_VALUE.getMillis());
+
batch.add(element.getValue());
// Blind add is supported with combiningState
storedBatchSize.add(1L);
- // Add the timestamp back into minBufferedTs as it might be cleared by
flushBatch above.
minBufferedTs.add(elementTs.getMillis());
final long num = storedBatchSize.read();
+ long currentMinBufferedTs = minBufferedTs.read();
// If this is the first element in the batch or if the timer's output
timestamp needs
// modifying, then set a timer.
- long oldOutputTs =
- MoreObjects.firstNonNull(
- minBufferedTs.read(),
BoundedWindow.TIMESTAMP_MAX_VALUE.getMillis());
- boolean needsNewTimer = num == 1 || minBufferedTs.read() != oldOutputTs;
+ boolean needsNewTimer = num == 1 || currentMinBufferedTs < oldOutputTs;
Review Comment:
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.
--
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]
