acrites commented on code in PR #40754:
URL: https://github.com/apache/beam/pull/40754#discussion_r4449055495
##########
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:
I guess the argument is that any time the output timestamp changes, we need
to set a new timer to respect that change. I'll change it back to `!=`.
--
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]
