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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org