Hi Alieh, Thanks for taking the time to review the KIP!
Regarding AS1: You make a good point about equivalency with Flink. I was looking into FLIP-11 ( https://cwiki.apache.org/confluence/spaces/FLINK/pages/65868406/FLIP-11+Table+API+Stream+Aggregations) and the original proposal for flink did include support for FOLLOWING. I've briefly looked into the jira ticket discussions and mailing list, but couldn't find the reason why it was omitted in the end. Do you by any chance know the reason for this? The current design already buffers the events, so performance/implementation-wise supporting FOLLOWING is straightforward. If events are arriving out of order, it might also be a useful feature in anomaly detection systems etc. Given the semantics, in a fully in-order stream, after lookups always returns zero records — results only appear when earlier-timestamped records arrive after higher-timestamped ones have already been buffered. This might be confusing for users. To maintain consistency with flink and avoid user confusion, I've moved the `after` from the constructor of the ranges into an optional builder function `.withLookAhead()`. Does this seem like a good option? Or do you see further reason to completely drop the feature? Regarding AS2: Thank you for mentioning the naming collision, this is indeed problematic. Ideally, to decouple the aggregators from the ranges, I'd like to give the RangeAggregator access to the Iterable<Record<K,V>>. In order to do that, I'd need the record metadata, the key and the value. Looking into the current available iterators: - WindowStoreIterator: extends KeyValueIterator<Long, V>. Losing the key might actually not be a bad thing, the responsibility for setting the key in the iterator switches from the overridable Range to the Range Processor in this case. However, it might complicate compatibility with KIP-1271: we are not able to retrieve the headers from the statestore anymore if we prefer so. Furthermore, it has some functions like `peekNextKey` which are not available from the aggregators, but would need to be overridden just for type compatibility. From a domain standpoint, it's also weird to have ranges construct WindowStoreIterators, and leaks implementation details. - KeyValueIterator<K,V> has similar drawbacks, but doesn't lose the key. It's a better fit than the WindowStoreIterator if the headers are not important and the necessary peekNextKey implementation is not too much of an issue. For now, I've renamed the CloseableIterator in the KIP to RangedRecordIterator, to avoid the naming collision and avoid the peekNextKey implementation. If reuse is prefered over the drawbacks, using KeyValueIterator<K,V> works for me too. Let me know what your preference is. Thanks for the feedback! Hannes Op do 25 jun 2026 om 19:48 schreef Alieh Saeedi via dev < [email protected]>: > Hi Hannes, > > > Thanks for the KIP. This seems to be a gap worth closing: SQL's OVER (... > ROWS/RANGE BETWEEN ...) and Flink's over-aggregation have no DSL equivalent > in Kafka Streams, and ksqlDB doesn't offer it either. > > > I have a few higher-level comments and questions before getting into the > details. > > > AS1: Flink (which the KIP cites) supports only the backward PRECEDING ... > CURRENT ROW frame for streaming OVER aggregation and intentionally omits > FOLLOWING. Should we also drop it? > > > AS2: CloseableIterator: there is already > org.apache.kafka.common.utils.internals.CloseableIterator, and the Streams > store iterators (KeyValueIterator, WindowStoreIterator) are already > Closeable with close(). Could Range.fetch reuse an existing iterator type > rather than introducing a new public CloseableIterator with the same simple > name? If a Record-typed closeable iterator is genuinely needed, use a > non-colliding name and include a note explaining why the existing types > don't fit. > > > > Thanks > -Alieh > > > > On Wed, May 27, 2026 at 9:32 PM Hannes Buseyne <[email protected]> > wrote: > > > Hi all, > > > > I'd like to start a discussion on KIP-1352, which proposes adding range > > aggregations to the Kafka Streams DSL. > > The current windowing API is limited to fixed time intervals and > > incremental aggregations. This KIP introduces a Range abstraction that > > allows users to define flexible record-centric ranges (by event time or > > event count) and perform non-incremental aggregations over them, emitting > > results as a KStream. > > > > KIP page: > > > > > https://urldefense.com/v3/__https://cwiki.apache.org/confluence/display/KAFKA/KIP-1352*3A*Ranges*and*Range*Aggregations*in*the*Kafka*Streams*DSL__;JSsrKysrKysrKw!!Ayb5sqE7!t3zDQWQ6gAYrqYo_o-3JJJRSzDii-j5WOdFXRCB8ZhQ2PhfqcjA8z2XDW7rdkU0TYFMDBWtV_oNOKlTAayuvdsKvug$ > > > > Looking forward to your feedback > > Hannes Buseyne > > >
