kennknowles commented on code in PR #40754:
URL: https://github.com/apache/beam/pull/40754#discussion_r4404624809
##########
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:
If we had a bug where somehow `currentMinBufferedTs > oldOutputTs` by
accident, maybe it is better to have the bound instead of depending on `<`
being equivalent to `!=` due to context?
--
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]
