Status

Current state: Under Discussion

Discussion thread: link

JIRA: link

Pull Request: link

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

While the Kafka Streams DSL excels at temporal grouping through Tumbling, Hopping, and Sliding windows, these native constructs are fundamentally time-centric. They group records into absolute time buckets and output continuously updating KTable changelogs via incremental aggregation. Many real-world stream processing use cases require a more dynamic, event-centric context. Developers frequently need to evaluate a holistic 'range' of events relative to the arrival of a specific anchor record (e.g., 'the 3 events before and 2 events after this specific transaction'). Furthermore, these use cases often require the evaluation to be an immutable, point-in-time calculation rather than an infinitely updating KTable that triggers complex cascading updates when late data arrives.
Example use cases include calculating time-weighted moving averages in financial applications, building context-aware feature vectors for machine learning anomaly detection, and threshold monitoring. This event-centric, point-in-time pattern is a standard paradigm in stream processing (similar to the SQL OVER clause with ROWS/RANGE BETWEEN), and is natively supported by other frameworks like Apache Flink. It is a highly requested feature from the Kafka Streams community, e.g. (1), (2) and (3).
To bridge this gap, this KIP proposes introducing range-based query capabilities to the Kafka Streams DSL. This will empower developers to define dynamic, event-centric contexts and perform safe, non-incremental aggregations across those ranges.

Public Interfaces

The proposed changes to public interfaces include:

A new `Range` abstract class that both built-in and custom range definitions must extend, and two built-in implementations:

public abstract class Range<K, V> {

    private final long gracePeriodMs;

    /**
     * @param gracePeriodMs the grace period in milliseconds. Must not be negative.
     */
    protected Range(final long gracePeriodMs) {
        this.gracePeriodMs = gracePeriodMs;
    }

    /**
 	* Fetch the records that fall within this range for the given anchor record.
 	* The anchor record itself should be included in the returned iterator.
 	* Records should be ordered by their timestamps in ascending order.
 	* <p>
 	* The framework guarantees that this iterator will be safely closed after 
 	* the aggregation completes, preventing resource leaks.
 	*
 	* @param anchor the record that triggered the range evaluation
 	* @param store  the buffer store holding records for the anchor's group key
 	* @return a ranged record iterator of records that fall within the defined range
 	*/
	public abstract RangedRecordIterator<Record<K, V>> fetch(Record<K, V> anchor, ReadOnlyWindowStore<K, V> store);

    /**
     * @return the grace period in milliseconds. Records arriving after stream time has advanced
     * beyond the range's natural boundary plus this value will be dropped.
     */
    public long gracePeriodMs() {
        return gracePeriodMs;
    }

    /**
     * The minimum retention the buffer {@link WindowStore} must be configured with, excluding
     * the grace period. Implementations should return the oldest a record can be relative to an
     * anchor's timestamp and still fall within the range (e.g. {@code before} for
     * {@link EventTimeRange}, {@code maxTimeBefore} for {@link EventCountRange}).
     *
     * @return the range-specific retention in milliseconds, excluding grace period
     */
    protected abstract long rangeRetentionMs();

    /**
     * The minimum retention the buffer {@link WindowStore} must be configured with to correctly
     * serve this range, including the grace period. This is what {@code rangeOver()} validates
     * against the {@link Materialized} retention.
     *
     * @return the total required retention in milliseconds
     */
    public long retentionMs() {
        return rangeRetentionMs() + gracePeriodMs;
    }
}

public final class EventTimeRange<K, V> extends Range<K, V> {

    /**
	 * Create an {@link EventTimeRange} spanning {@code before} time prior to the anchor 
     * record's timestamp, capping the upper boundary strictly at the anchor record (Current Row).
     * Late-arriving records past the stream time watermark are dropped.
     * Use {@link #ofTimeBoundsAndGrace(Duration, Duration, Duration)} to tolerate late arrivals.
     * Use {@link #withMaxRecords(int)} to cap the number of records included.
     *
     * @param before the time before the anchor record's timestamp that defines the start of the range. Must not be negative.
     * @return a new {@link EventTimeRange} with no grace period
     * @throws IllegalArgumentException if the duration is negative or can't be represented as {@code long milliseconds}
     */
    public static <K, V> EventTimeRange<K, V> ofTimeBoundsWithNoGrace(final Duration before) {}

    /**
     * Create an {@link EventTimeRange} spanning {@code before} time prior to and {@code after} time following
     * the anchor record's timestamp, accepting late records up to {@code grace} beyond the range boundary.
     * Use {@link #withMaxRecords(int)} to cap the number of records included.
     *
     * @param before      the time before the anchor record's timestamp that defines the start of the range. Must not be negative.
     * @param grace       the grace period to tolerate late-arriving records. Must not be negative.
     * @return a new {@link EventTimeRange} with the specified grace period
     * @throws IllegalArgumentException if any duration is negative or can't be represented as {@code long milliseconds}
     */
    public static <K, V> EventTimeRange<K, V> ofTimeBoundsAndGrace(final Duration before, final Duration grace) {}

    /**
     * Cap the number of records included in the range. If more records fall within the time boundaries,
     * the newest records are dropped.
     *
     * @param maxRecords the maximum number of records to include. Must be positive.
     * @return this {@link EventTimeRange}
     * @throws IllegalArgumentException if {@code maxRecords} is not positive
     */
    public EventTimeRange<K, V> withMaxRecords(final int maxRecords) {}

    /**
     * Enable a forward-looking window over newer records that have already been 
     * buffered in the state store (useful for out-of-order anchor context resolution).
     *
     * @param after the time after the anchor record's timestamp to look forward. Must not be negative.
     * @return this {@link EventTimeRange} instance
	 * @throws IllegalArgumentException if the duration is negative or can't be represented as {@code long milliseconds}
     */
    public EventTimeRange<K, V> withLookAhead(final Duration after) {}
}

public final class EventCountRange<K, V> extends Range<K, V> {

    /**
	 * Create an {@link EventCountRange} including up to {@code before} records prior to 
     * the anchor record in event-time order, capping the upper boundary at the anchor record.
	 * Late-arriving records past the stream time watermark are dropped.
     * Use {@link #ofCountBoundsAndGrace(int, int, Duration, Duration)} to tolerate late arrivals.
     * Use {@link #withLookAhead(int)} to include lookahead for late arriving records
     * Use {@link #withMaxTimeAfter(Duration)} to add an optional time ceiling on the forward direction.
     *
     * @param before        the number of records before the anchor record to include. Must not be negative.
     * @param maxTimeBefore the maximum time before the anchor's timestamp to look back. Must not be negative.
     * @return a new {@link EventCountRange} with no grace period
     * @throws IllegalArgumentException if the count is negative or the duration can't be represented as {@code long milliseconds}
     */
    public static <K, V> EventCountRange<K, V> ofCountBoundsWithNoGrace(final int before, final Duration maxTimeBefore) {}

    /**
     * Create an {@link EventCountRange} including a count of {@code before} records prior to records following,
     * the anchor record in event-time order, accepting late records up to {@code grace} beyond the range boundary.
     * {@code maxTimeBefore} sets a required time floor: records older than {@code anchor.timestamp - maxTimeBefore}
     * are excluded regardless of count.
     * Use {@link #withMaxTimeAfter(Duration)} to add an optional time ceiling on the forward direction.
     *
     * @param before        the number of records before the anchor record to include. Must not be negative.
     * @param maxTimeBefore the maximum time before the anchor's timestamp to look back. Must not be negative.
     * @param grace         the grace period to tolerate late-arriving records. Must not be negative.
     * @return a new {@link EventCountRange} with the specified grace period
     * @throws IllegalArgumentException if any count is negative or any duration can't be represented as {@code long milliseconds}
     */
    public static <K, V> EventCountRange<K, V> ofCountBoundsAndGrace(final int before, final Duration maxTimeBefore, final Duration grace) {}

    /**
     * Set an optional time ceiling on the forward direction. Records newer than
     * {@code anchor.timestamp + maxTimeAfter} are excluded regardless of count.
     *
     * @param maxTimeAfter the maximum time after the anchor's timestamp to look forward. Must not be negative.
     * @return this {@link EventCountRange}
     * @throws IllegalArgumentException if the duration is negative or can't be represented as {@code long milliseconds}
     */
    public EventCountRange<K, V> withMaxTimeAfter(final Duration maxTimeAfter) {}

    /**
     * Enable a forward-looking window over a count of newer records that have already 
     * been buffered in the shared state store (useful for out-of-order anchor context resolution).
     *
     * @param after the number of records after the anchor to include. Must not be negative.
     * @return this {@link EventCountRange} instance
	 * @throws IllegalArgumentException if the count is negative
     */
    public EventCountRange<K, V> withLookAhead(final int after) {}
}

A new public interface extending java.util.Iterator and java.lang.AutoCloseable is introduced to manage resource-backed iterators safely without forcing users to handle checked exceptions.

package org.apache.kafka.streams.kstream;

import java.util.Iterator;

/**
 * An {@link Iterator} that holds resources (e.g. a RocksDB cursor) and must be closed after use.
 * The {@link #close()} method does not throw a checked exception, making it safe for use in
 * try-with-resources blocks without requiring a catch clause.
 *
 * <p>Implementations must be idempotent: calling {@link #close()} more than once must be safe.
 *
 * @param <T> the type of elements returned by this iterator
 */
public interface RangedRecordIterator<T> extends Iterator<T>, AutoCloseable {
    @Override
    void close();
}

Two new methods `rangeOver` in the `KGroupedStream` interface:

public interface KGroupedStream<K, V> {
    // existing methods...
	
	/**
     * Create a new {@link RangedKStream} instance with a default internal buffer store.
     * The store is auto-named, uses the Serdes of this grouped stream, and its retention is set to
     * {@link Range#retentionMs()} — the minimum required to serve the range. Use
     * {@link #rangeOver(Range, Materialized)} to name the store, override Serdes, or increase retention.
     * <p>
     * Records with {@code null} key or {@code null} value are dropped and not written to the buffer store.
     * The {@code dropped-records-total} metric is incremented for each dropped record.
     *
     * @param range the range definition, determining which records are included for each anchor record
     * @return an instance of {@link RangedKStream}
     */
    RangedKStream<K, V> rangeOver(final Range<? super K, ? super V> range);

    /**
     * Create a new {@link RangedKStream} instance that can be used to perform ranged aggregations on the grouped stream.
     * The range of the aggregation is defined by the provided {@link Range} instance.
     * Built-in implementations are provided via {@link EventTimeRange} and {@link EventCountRange}.
     * Custom implementations of {@link Range} can also be provided.
     * <p>
     * The {@code materialized} parameter configures the underlying buffer store, which holds the raw records
     * used by the range definition to fetch records on each trigger. This store is materialized at this step,
     * not at aggregation time, because the aggregation result is emitted as a {@link KStream} and not persisted.
     * <p>
     * Records with {@code null} key or {@code null} value are dropped and not written to the buffer store.
     * The {@code dropped-records-total} metric is incremented for each dropped record.
     *
     * @param range        the range definition, determining which records are included for each anchor record
     * @param materialized the configuration for the underlying buffer state store
     * @return an instance of {@link RangedKStream}
     * @throws IllegalArgumentException if the retention period specified in {@code materialized} is smaller than
     * {@link Range#retentionMs()}.
     */
    RangedKStream<K, V> rangeOver(
        final Range<? super K, ? super V> range,
        final Materialized<K, V, WindowStore<Bytes, byte[]>> materialized
    );
}

And a RangedKStream interface:

public interface RangedKStream<K, V> {
    /**
     * Perform an aggregation on the records in the range defined for this stream. 
     * The aggregation will be triggered for each incoming record and will include all records that fall within the defined range of that record.
     *
     * @param aggregator the aggregator function to apply to the records in the range
     * @param <VR> the type of the aggregated value
     * @return a {@link KStream} containing the aggregated results for each unmodified key, with the output record's timestamp inherited from the anchor record.
     */
    <VR> KStream<K, VR> aggregate(final RangeAggregator<K, V, VR> aggregator);

    /**
     * Count the number of records in this range by the grouped key and defined range.
     *
     * @return a {@link KStream} that contains records with unmodified keys and {@link Long} values
     * that represent the current count of records for the defined range
     */
    KStream<K, Long> count();

}

And a new functional interface for the range aggregator:

@FunctionalInterface
public interface RangeAggregator<K, V, VR> {
    /**
     * Apply the aggregation logic to the records in the defined range for a given key and timestamp.
     * @param anchor the record that triggered the aggregation
     * @param rangeRecords a read-only iterable of records that fall within the defined range of the anchor record, including the anchor record itself. The records are ordered by their timestamps in ascending order.
     * @return the result of the aggregation
     */
    VR apply(Record<K,V> anchor, Iterable<Record<K,V>> rangeRecords);
}

Proposed Changes

This KIP introduces the Range<K, V> abstract class as the primary extension point for defining range boundaries. It provides two built-in implementations:

• EventTimeRange: Defines ranges based strictly on event-time durations (e.g., N seconds before and after the anchor).

• EventCountRange: Defines ranges based on strict record counts (e.g., N records before and after), bounded by a required maximum look-back time to ensure safe state store retention.

To align with modern Kafka Streams API standards (KIP-633), the grace period for late-arriving records is enforced strictly via static factory methods and constructors. Furthermore, users can pass custom Range subclasses into rangeOver() to implement arbitrary fetching logic beyond the built-in types.

Physical topology & state sharing

A key architectural design of this KIP is that the buffer store is materialized at rangeOver() time, not at aggregation time.

In the physical topology, rangeOver() injects a dedicated RangeStoreProcessor that owns the WindowStore. Each subsequent call to .aggregate() on the resulting RangedKStream introduces an independent RangeAggregateProcessor as a child node. Because the state store is wired using low-level topology configurations, multiple .aggregate() calls on the same range seamlessly share a single underlying RocksDB instance, completely eliminating data duplication and redundant storage overhead.

Iterator Lifecycle Management: To ensure optimal O(1) memory efficiency without exposing users to resource leaks, Range.fetch() returns a RangedRecordIterator. The RangeAggregateProcessor wraps this cursor in a lazy Iterable for the user's RangeAggregator, and mathematically guarantees the cursor is closed via a try-with-resources block after the aggregation completes—even if the user exits the loop early or an exception is thrown.

Processing Flow

The sequence diagram below illustrates the lifecycle of a single incoming record through this topology:

1. Buffer Phase: The RangeStore Processor receives the incoming record, evaluates it against the defined grace period, and drops it if it is late. Valid records are persisted to the shared WindowStore and immediately forwarded downstream.

2. Fetch Phase: The RangeAggregate Processor receives the forwarded record (which now acts as the "anchor"). It delegates the fetch operation to the Range implementation, which queries the shared state store to build the context boundaries.

3. Aggregate Phase: The anchor record and the fetched holistic context (Iterable<Record<K, V>>) are passed to the RangeAggregator. The stateless calculation is performed, and the final result is forwarded downstream as a pure, append-only KStream record.

4. Cleanup Phase: The framework mathematically guarantees that the underlying RocksDB cursor is safely closed via a try-with-resources block, even if the user's aggregator exits the loop early or throws an exception.

Usage

EventTimeRange — compute a rolling aggregation over a 20-second window around each sensor reading.

KStream<String, SensorReading> readings = builder.stream("sensors");

RangedKStream<String, SensorReading> ranged = readings
    .groupByKey()
    .rangeOver(
        EventTimeRange.ofTimeBoundsAndGrace(Duration.ofSeconds(20), Duration.ofSeconds(5))
            .withLookAhead(Duration.ofSeconds(10)), // Explicit look-ahead opt-in
        Materialized.<String, SensorReading, WindowStore<Bytes, byte[]>>as("sensor-buffer")
                    .withRetention(Duration.ofSeconds(35))
    );

KStream<String, Double> movingAverage = ranged.aggregate((anchor, rangeRecords) -> {
    double sum = 0;
    int count = 0;
    for (Record<String, SensorReading> r : rangeRecords) {
        sum += r.value().reading();
        count++;
    }
    return sum / count;
});

// Note that multiple aggregations on 1 buffer are possible

KStream<String, Double> median = ranged.aggregate((anchor, rangeRecords) -> {
    final List<Double> values = new ArrayList<>();
    for (Record<String, SensorReading> r : rangeRecords) {
        values.add(r.value().reading());
    }
    Collections.sort(values);
    final int size = values.size();
    return size % 2 == 0
        ? (values.get(size / 2 - 1) + values.get(size / 2)) / 2.0
        : values.get(size / 2);
});

EventCountRange — include the 3 events before and 0 events after each record, looking back at most 1 hour:

KStream<String, Long> rangeCounts = readings
    .groupByKey()
    .rangeOver(
        EventCountRange.ofCountBoundsAndGrace(3, Duration.ofHours(1), Duration.ofSeconds(5))
            .withLookAhead(2), // Explicitly look forward at up to 2 buffered records
        Materialized.<String, SensorReading, WindowStore<Bytes, byte[]>>as("sensor-count-buffer")
                    .withRetention(Duration.ofHours(1).plusSeconds(5))
    )
    .count();

Custom Range — subclass `Range` to implement arbitrary fetch logic. This example groups all records that fall within the same calendar hour as the anchor, regardless of how far apart they are from the anchor itself. This kind of calendar-aligned boundary is not possible with `EventTimeRange`, which is always relative to the anchor's timestamp:

public class HourAlignedRange extends Range<String, SensorReading> {
    private static final long HOUR_MS = Duration.ofHours(1).toMillis();

    public HourAlignedRange(final Duration grace) {
        super(validateMillisecondDuration(grace, "grace"));
    }

    @Override
	public RangedRecordIterator<Record<String, SensorReading>> fetch(
        	final Record<String, SensorReading> anchor,
        	final ReadOnlyWindowStore<String, SensorReading> store) {
    
    	final long hourStart = anchor.timestamp() - (anchor.timestamp() % HOUR_MS);
    	final long hourEnd = hourStart + HOUR_MS;
    
    	// Fetch the raw RocksDB iterator
    	final WindowStoreIterator<SensorReading> storeIterator = 
            store.fetch(anchor.key(), Instant.ofEpochMilli(hourStart), 	Instant.ofEpochMilli(hourEnd));
            
    	// Return a lazy RangedRecordIterator that maps the KeyValue to a Record
    	return new RangedRecordIterator<Record<String, SensorReading>>() {
        	@Override
        	public boolean hasNext() { return storeIterator.hasNext(); }

        	@Override
        	public Record<String, SensorReading> next() {
            	KeyValue<Long, SensorReading> kv = storeIterator.next();
            	return new Record<>(anchor.key(), kv.value, kv.key);
        	}

        	@Override
        	public void close() { storeIterator.close(); }
    	};
	}

    @Override
    protected long rangeRetentionMs() { return HOUR_MS; }
}

The following illustrations demonstrate how the two built-in range types behave:

EventTimeRange

The diagram above illustrates four events belonging to the same group key. The RangedKStream buffers these events in the shared state store. To support multiple events with the exact same timestamp, the internal WindowStore is configured with retainDuplicates = true, which appends a sequence number to the composite key.

After processing the first three events (A, C, D) in order, the state store looks like this:

When an out-of-order or late-arriving event (B) arrives at timestamp 50, the processor immediately writes it to the store:

To evaluate the range for anchor B (assuming a configuration of before = 20 and lookahead = 30), the processor executes a single highly efficient range query: store.fetch(key, 30, 80). This returns an iterator containing A, B, and C.

The "Forward-Looking" Constraint & maxRecords: Because aggregations are emitted immediately to a KStream upon record arrival, an in-order record will never see "future" records (since they haven't arrived yet). Therefore, the after parameter strictly serves to capture context for out-of-order or late-arriving anchors evaluating against already-buffered newer records. If the fetched range exceeds the optional maxRecords parameter, the newest records in the iterator are dropped. This guarantees that unneeded records are never deserialized, saving CPU cycles.

EventCountRange

The diagram above illustrates the same late-arrival scenario, this time applying an EventCountRange with before = 1 and lookahead = 1.

When the late-arriving anchor event (B at timestamp 50) arrives, it is written to the store. To resolve the count-based boundaries, the RangedKStream executes a two-pronged fetch strategy:

1. The Backward Fetch: It executes a store.backwardFetch() from the anchor down to anchor.timestamp - maxTimeBefore. This iterator yields records descending in time, which the processor limits to the before count (yielding event A).

2. The Forward Fetch: It executes a standard store.fetch() from the anchor up to the current stream time (or maxTimeAfter if configured). This iterator yields records ascending in time, which the processor limits to the after count (yielding event C).

To ensure the final Iterable presented to the RangeAggregator remains in strict chronological ascending order, the processor reverses the bounded backward-fetch results in memory (an O(before) operation) and concatenates it with the anchor and the forward-fetch iterator.

Boundary Non-Determinism: Just as with EventTimeRange, the after count strictly applies to out-of-order or late-arriving records looking at newer, already-buffered data. Note on duplicates: Because retainDuplicates = true uses physical sequence numbers, if multiple records share the exact same timestamp at the exact cut-off boundary of the before or after count, the specific records included in the iterator are technically non-deterministic, aligning with standard Kafka Streams windowing behavior.

Retention Period

To guarantee that context is safely preserved for out-of-order and late-arriving records, the buffer WindowStore must be configured with a retention period of at least Range.retentionMs(), which is computed as rangeRetentionMs() + gracePeriodMs(). The rangeOver() method internally validates this against the user-provided Materialized configuration and throws an IllegalArgumentException during topology initialization if the retention is too small.

The rangeRetentionMs() bounds for the built-in types are:

• EventTimeRange: The before duration.

• EventCountRange: The maxTimeBefore duration.

Custom subclasses must implement rangeRetentionMs() to mathematically return the absolute maximum age a record can have relative to an anchor's timestamp and still legally fall within the custom range.

Extendability

The Range<K, V> abstract class serves as the primary extension point. Users can subclass it to implement custom range logic and pass the result directly to rangeOver().

The following concepts are explicitly out of scope for this KIP but are natural candidates for future follow-up work:

• Additional built-in range types: Other range definitions beyond EventTimeRange and EventCountRange can be added as further subclasses of Range in the future without requiring any overarching API changes.

• rangeOver on plain KStream: Ranges without prior grouping would require a completely different buffering strategy (such as a global or partition-wide state store) since there is no specific group key to tightly scope the physical store.

• rangeOver on CoGroupedKStream: Co-grouped ranges would require a highly complex shared multi-stream buffer store and a completely revised aggregator interface to handle records from heterogeneous value types. This is deferred accordingly.

Performance Considerations

Because range aggregations are dynamically evaluated upon the arrival of every anchor record, performance will scale relative to the density and size of the fetched ranges.

• Memory Footprint: The API strictly bounds memory usage by passing a lazy Iterable (backed by a RangedRecordIterator) to the RangeAggregator. This ensures that processing exceptionally large ranges does not load the entire dataset into the JVM heap, preventing OutOfMemory (OOM) errors and Garbage Collection (GC) spikes.

• CPU & I/O Overhead: Evaluating ranges requires querying the underlying state store. For an EventTimeRange, exceptionally wide time boundaries combined with a high (or omitted) maxRecords cap will increase RocksDB read operations and deserialization overhead per anchor record.

• Multiple Aggregations & Query Efficiency: Because a RangedKStream allows multiple independent aggregations to be chained off a single shared buffer store, each subsequent call to .aggregate() registers a distinct RangeAggregateProcessor child node in the physical topology. Consequently, each child processor independently invokes Range.fetch() for a given anchor record. While this results in multiple range queries for the same anchor, the execution overhead is heavily minimized:

  • Depth-First Execution: Kafka Streams executes topologies sequentially using a depth-first strategy. The exact same anchor record is forwarded to each aggregate child processor in immediate, microsecond-level succession.

  • Cache Optimization: Because subsequent lookups occur almost simultaneously for the identical key and boundaries, the data blocks are pinned directly in the RocksDB Block Cache or the OS Page Cache. Sibling scans read strictly out of RAM, completely avoiding physical disk I/O thrashing.

  • Memory Safety Priority: This architecture explicitly favors a guaranteed O(1) JVM heap footprint over redundant seeks. Eagerly collecting records into an upstream list to forward downstream would impose a severe garbage collection and heap penalty on all users, even those with only a single aggregator.

• Tuning & Best Practices: * Users operating on high-throughput streams are highly encouraged to tune maxRecords (for time ranges) and maxTimeBefore (for count ranges) to the tightest business requirements possible to minimize unnecessary deserialization.

  • For exceptionally dense or massive ranges where users are hyper-sensitive to iteration overhead across multiple metrics, the API inherently supports single-pass query efficiency. Users can aggregate their data into a single composite Java object within a single .aggregate() call block and split the streams downstream via .mapValues().

Compatibility, Deprecation, and Migration Plan

The proposed changes are backward compatible as they introduce new interfaces and methods without modifying existing ones.
There is, however possible confusion with the existing windowing functionality, since the method names for aggregation are the same but the semantics and return types are different. This distinction needs to be emphasized in the documentation.

Test Plan

Range Definition Unit Tests

Mirrors SessionWindowsTest and SlidingWindowsTest.

• Parameter Validation: Verify that before, after, and maxTimeBefore are not negative; maxRecords is strictly positive; and grace is not negative.

• Duration Conversion: IllegalArgumentException is thrown for durations that cannot be safely represented as milliseconds.

• State Resolution: gracePeriodMs() correctly returns the value passed during construction.

• Retention Math: retentionMs() correctly equals rangeRetentionMs() + gracePeriodMs().

• Built-in Bounds: EventTimeRange.rangeRetentionMs() equals before in milliseconds. EventCountRange.rangeRetentionMs() equals maxTimeBefore in milliseconds.

• Fluent Chaining: withMaxRecords() and withMaxTimeAfter() correctly return their concrete subtypes to maintain fluent chaining.

• Builder Validation: rangeOver() throws IllegalArgumentException if the Materialized retention is smaller than range.retentionMs().

• Store Validation: rangeOver() throws IllegalArgumentException if configured with a custom WindowBytesStoreSupplier where retainDuplicates() returns false.

• Lambda Support: Verify that RangeAggregator can be cleanly expressed as a Java lambda expression.

Topology & Integration Tests

Mirrors TimeWindowedKStreamIntegrationTest.

• EventTimeRange End-to-End: Records produce the correct aggregated output per anchor; output is written to a downstream Kafka topic with the unmodified key, aggregated value, and the anchor's original timestamp.

• Grace Period & Late Data: Late records arriving within the configured grace period produce updated output; records arriving past the grace period are correctly dropped.

• Range Capping: EventTimeRange with withMaxRecords strictly caps the output and drops the newest records when the range exceeds the configured limit.

• EventCountRange End-to-End: Correct before/after counts are evaluated and included per anchor.

• Count Time Floors/Ceilings: EventCountRange with maxTimeBefore strictly excludes records beyond the time floor regardless of count. EventCountRange with withMaxTimeAfter strictly excludes records newer than anchor.timestamp + maxTimeAfter regardless of count.

• Iterator Lifecycle Safety: Verify that the underlying RangedRecordIterator is successfully closed by the framework processor after a successful aggregation loop, and explicitly verify it is immediately closed via try-with-resources if the user-defined RangeAggregator lambda throws an unexpected RuntimeException mid-flight.

• Null Handling: Records with a null key or null value are dropped, and the dropped-records-total metric is successfully incremented.

• count(): count() operator end-to-end test emits the correct long value per anchor.

• Custom Subclasses: A custom Range subclass is correctly integrated, materialized, and fetched via rangeOver().

• State Restoration: After a topology restart, the shared buffer store is successfully restored from its changelog topic and range aggregations resume correctly.

• Caching Behavior: Output is verified to be identical with caching enabled vs. disabled. This ensures that Range.fetch() correctly reads newly written records in the exact same processing step, regardless of whether they have been flushed to the physical RocksDB store yet.

Rejected Alternatives

• Naming (bufferedBy vs. rangeOver): Using .bufferedBy() alongside EventTimeBuffer or EventCountBuffer was rejected. The term "buffer" is already heavily overloaded within Kafka and Kafka Streams internals (e.g., Suppress buffers, RecordCollector buffers). Furthermore, the term "range" tightly aligns with industry-standard streaming SQL semantics (such as Flink's RANGE OVER or standard SQL ROWS BETWEEN).

• Implementation as a New Window Type: Implementing this directly within the existing .windowedBy() API was rejected for two structural reasons:

  • Semantics: Range aggregations fundamentally emit an append-only KStream and do not receive retroactive updates when late data arrives. Standard window aggregations are incremental and result in a continuously updating KTable.

  • Materialization Lifecycle: Standard windows only require state store materialization at the aggregation step. Ranges require a materialized buffer store immediately upon definition to physically execute the look-back/look-forward queries on the raw records, even before an aggregation is applied.

• Retroactive Updates for Past Ranges (Emitting a KTable): When late data arrives, we do not emit retroactive updates for previously evaluated ranges. Returning an updating KTable instead of an append-only KStream was rejected because overlapping event-centric ranges mean a single late-arriving record mathematically alters the context of N surrounding anchors. Emitting retroactive updates for all affected past anchors would cause exponential write amplification and severe RocksDB read-thrashing.