vcrfxia commented on code in PR #13564:
URL: https://github.com/apache/kafka/pull/13564#discussion_r1171926005
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamAggregate.java:
##########
@@ -118,10 +120,13 @@ public void process(final Record<KIn, VIn> record) {
newAgg = aggregator.apply(record.key(), record.value(), oldAgg);
- store.put(record.key(), newAgg, newTimestamp);
- tupleForwarder.maybeForward(
- record.withValue(new Change<>(newAgg, sendOldValues ? oldAgg :
null))
- .withTimestamp(newTimestamp));
+ final long putReturnCode = store.put(record.key(), newAgg,
newTimestamp);
Review Comment:
Yes, we do not attempt to "fix" older aggregation results. I can add a note
in the KIP.
See my comment above about whether we can actually guarantee that result
timestamps are nondecreasing:
https://github.com/apache/kafka/pull/13564#discussion_r1171925337 Same question
applies for both table aggregations and stream aggregations.
--
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]
