KAFKA-20704: IQv2 TimestampedRangeWithHeadersQuery for headers-aware key-value stores (KIP-1356) by Jess668 · Pull Request #22770 · apache/kafka

Jess668 · 2026-07-06T21:36:00Z

Implements TimestampedRangeWithHeadersQuery, the second IQv2 query type of KIP-1356, whose result is an iterator of records that each carry key, value, timestamp, and headers. The result type is the ReadOnlyRecordIterator<K, V> — a Closeable iterator that yields ReadOnlyRecord<K, V> elements directly.

Dependencies

This work is stacked on, and should be reviewed/merged after:

ReadOnlyRecord PR#22677 — already in trunk.
ReadOnlyRecordIterator PR#22729 — already in trunk.
TimestampedKeyWithHeadersQuery PR#22666 — already in trunk.

All three have landed; this branch has been rebased onto trunk and contains only the range commits.

Summary

TimestampedRangeWithHeadersQuery<K, V> implements Query<ReadOnlyRecordIterator<K, V>>
(@Evolving / @InterfaceAudience.Public). Mirrors TimestampedRangeQuery: static factories withRange, withUpperBound,
withLowerBound, withNoBounds; order builders withDescendingKeys / withAscendingKeys;
accessors lowerBound(), upperBound(), resultOrder(). Same bound/order semantics as
TimestampedRangeQuery.
MeteredTimestampedKeyValueStoreWithHeaders registers a TimestampedRangeWithHeadersQuery
handler and returns a ReadOnlyRecordIterator whose elements are built as immutable
ReadOnlyRecords (headers frozen via setReadOnly()). The bounds+order → raw RangeQuery
construction shared by the three range handlers is factored into a rawRangeQuery(...) helper.
Native store enablement: RocksDBTimestampedStoreWithHeaders drops its
scope-to-KeyQuery-only query() override, falling back to the inherited RocksDBStore handling,
which serves RangeQuery. Previously the native header store returned UNKNOWN_QUERY_TYPE for
range queries.
num-open-iterators metric fix: the IQ range iterators (QueryIterator, used by
RangeQuery/TimestampedRangeQuery, and the new ReadOnlyRecordIterator for this query)
registered in openIterators but never updated numOpenIterators, so they were invisible to the
num-open-iterators gauge. Both now increment on open / decrement on close, matching the base
MeteredKeyValueStoreIterator.

Testing

Unit — MeteredTimestampedKeyValueStoreWithHeadersTest: bounds (as serialized keys) and
ascending/descending order are propagated to the underlying raw RangeQuery.
Builder / end-to-end — TimestampedKeyValueStoreBuilderWithHeadersTest: RangeQuery on the
native store; TimestampedRangeWithHeadersQuery returns exact headers on a header-persisting
store, empty headers on the adapter store, throws on a plain/legacy store, throws on a negative
stored timestamp; execution-info collect/no-collect; identical range results across native- and
adapter-built stores.
Native store — RocksDBTimestampedStoreWithHeadersTest: RangeQuery is now handled (was
UNKNOWN_QUERY_TYPE).
Integration — IQv2HeadersStoreIntegrationTest: TimestampedRangeWithHeadersQuery over a
running topology returns each record's correct key/value/timestamp/headers across order and bound
variants; UNKNOWN_QUERY_TYPE against a non-headers store; failure against a plain supplier.

Reviewers: Alieh Saeedi asaeedi@confluent.io

- Add the headers-aware range query TimestampedRangeWithHeadersQuery, returning a ReadOnlyRecordIterator of ReadOnlyRecord (key/value/timestamp/headers). - MeteredTimestampedKeyValueStoreWithHeaders handles it by forwarding a raw RangeQuery to the wrapped store and adapting each entry into a frozen-header ReadOnlyRecord; an entry whose stored timestamp is negative cannot be represented as a ReadOnlyRecord, so advancing the iterator throws StreamsException. - Also drop the query() override on RocksDBTimestampedStoreWithHeaders so the native header store serves the basic IQv2 queries via the inherited RocksDBStore handling.

- TimestampedKeyValueStoreBuilderWithHeadersTest: end-to-end build-path coverage (native/in-memory return headers; adapter returns empty headers; plain-legacy and negative timestamps throw while iterating), execution-info collection, and native/adapter parity. - MeteredTimestampedKeyValueStoreWithHeadersTest: unit coverage of the metered handler forwarding the bounds and result order onto the raw RangeQuery. - RocksDBTimestampedStoreWithHeadersTest: the native store now serves RangeQuery. - TimestampedRangeWithHeadersQueryIntegrationTest: IQv2 integration coverage.

…ntion - Query with PositionBound.at(INPUT_POSITION) + iqv2WaitForResult

aliehsaeedii

Thanks @Jess668. The PR looks good, but I still spent some time catching a few final issues to get the code ready for merge.

aliehsaeedii · 2026-07-09T21:35:34Z

            this.startNs = time.nanoseconds();
            this.startTimestampMs = time.milliseconds();
            this.returnPlainValue = returnPlainValue;
+            numOpenIterators.increment();


This metric fix ships without a regression test. The only num-open-iterators test drives all(), which already tracked the gauge on trunk — nothing asserts the gauge moves for the range/IQ query iterator you fix here (or the new ReadOnlyRecordIterator). Add a 0→1→0 check over a range query iterator.

aliehsaeedii · 2026-07-09T21:35:34Z

+            final Headers headers = valueTimestampHeaders.headers();
+            final K key = deserializeKey(keyValue.key.get(), headers);
+            if (valueTimestampHeaders.timestamp() < 0) {
+                throw new StreamsException(


This throw leaves the iterator open. For a plain/NO_TIMESTAMP store the throw is an expected outcome of normal iteration, so a caller that catches it and abandons the iterator leaks the underlying RocksDB iterator and permanently inflates num-open-iterators. Worth documenting that the caller must close in a finally/try-with-resources.

aliehsaeedii · 2026-07-09T21:35:34Z

                final long duration = time.nanoseconds() - startNs;
                sensor.record(duration);
                iteratorDurationSensor.record(duration);
+                numOpenIterators.decrement();


The sibling MeteredTimestampedKeyValueStore has the same gauge bug this PR fixes here: its MeteredTimestampedKeyValueStoreIterator does openIterators.add/remove but never numOpenIterators.increment()/decrement(), so num-open-iterators under-counts for plain timestamped stores. Could you please open a jira ticket for that too? Maybe not related to your KIP, but just as a bug.

aliehsaeedii · 2026-07-09T21:35:34Z

+     * entry is read, such an entry cannot be surfaced as a query-level failure; it is instead reported
+     * by throwing a {@link StreamsException} while advancing the iterator.
+     */
+    private class MeteredTimestampedKeyValueStoreWithHeadersReadOnlyRecordIterator


This is now the third inner iterator in this file repeating the same scaffolding — fields, increment()/openIterators.add() in the ctor, and sensor-record + decrement()/remove() in close(). Consider extracting a shared abstract metered-iterator base so each subclass only implements next(). Optional (it matches the current idiom), but the duplication is growing.

aliehsaeedii · 2026-07-09T21:35:34Z

+            result = (QueryResult<R>) typedQueryResult;
+        } else {
+            // the generic type doesn't matter, since failed queries have no result set.
+            result = (QueryResult<R>) rawResult;


This failure branch (wrapped query returns a failure) has no test — every range-with-headers test asserts isSuccess(). Minor, but a failed-rawResult case would close it (the sibling handlers share the same untested else).

aliehsaeedii · 2026-07-09T21:35:35Z

+
+            assertTrue(result.isSuccess(), "The range query itself succeeds; the failure surfaces while iterating");
+            try (ReadOnlyRecordIterator<String, String> iterator = result.getResult()) {
+                assertThrows(StreamsException.class, iterator::next,


The negative-timestamp StreamsException is asserted only by type, never by message. If the message text matters, assert a fragment of it; otherwise fine to leave. Minor.

aliehsaeedii · 2026-07-09T21:35:35Z

-        IntegrationTestUtils.startApplicationAndWaitUntilRunning(kafkaStreams);
+    @Test
+    public void shouldHandleTimestampedRangeWithHeadersQuery() throws Exception {
+        // Caching disabled: a range query reads the underlying store directly (it never consults the


No integration test runs a range query with caching ENABLED. Since CachingKeyValueStoreWithHeaders.query() forwards straight to the store (bypassing the cache), a caching-enabled range query won't see writes not yet flushed — worth a test that locks in that behavior (the point query has its cache-hit counterpart).

aliehsaeedii · 2026-07-09T21:35:35Z

+
+        // Bounded ranges (inclusive on both ends); same per-element checks as the full scans.
+        // withRange(2, 3) -> keys 2, 3.
+        final List<ReadOnlyRecord<Integer, String>> range = rangeQuery(TimestampedRangeWithHeadersQuery.withRange(2, 3));


The bounded ranges (withRange/withLowerBound/withUpperBound) are only exercised in the default order. Add one descending bounded range so bounds + DESCENDING is covered end-to-end (only the full scan tests descending today).

aliehsaeedii · 2026-07-09T21:35:35Z

+
+        // withUpperBound(2) -> keys 1, 2.
+        final List<ReadOnlyRecord<Integer, String>> upperBounded =
+            rangeQuery(TimestampedRangeWithHeadersQuery.withUpperBound(2));


No range test asserts an empty result (a bound matching no keys). A quick empty-range case would cover the no-match path.

aliehsaeedii · 2026-07-09T21:35:35Z

+            PlainStoreWriterProcessor::new);
+    }
+
+    private void startStreamsWithPlainSupplierStore() throws Exception {


The ADAPTER config isn't covered end-to-end: a WithHeaders builder over a plain timestamped supplier (persistentTimestampedKeyValueStore) keeps timestamps but drops headers, so a range query returns empty (not null) headers — a distinct outcome from the three helpers here, tested only at the store level. Consider a startStreamsWithAdapterStore helper + a range test asserting empty headers.

github-actions Bot added triage PRs from the community streams labels Jul 6, 2026

aliehsaeedii reviewed Jul 7, 2026

View reviewed changes

Comment thread streams/src/main/java/org/apache/kafka/streams/query/TimestampedRangeWithHeadersQuery.java

github-actions Bot removed the triage PRs from the community label Jul 8, 2026

Jess668 force-pushed the kip-1356-timestamped-range-with-headers-query branch from a3efe3d to 8bd56d9 Compare July 8, 2026 14:52

Jess668 added 12 commits July 9, 2026 10:21

Add ascending-order range test and restore test indentation (KIP-1356)

7754774

Align range-query integration test with the IQv2 position-bound conve…

f57679b

…ntion - Query with PositionBound.at(INPUT_POSITION) + iqv2WaitForResult

Move range-query integration tests into IQv2HeadersStoreIntegrationTest

363774c

Keep point-query BuilderTest methods in place to reduce diff noise

d83537f

Extract shared raw RangeQuery construction into helper

e273e59

Track ReadOnlyRecordIterator in num-open-iterators metric

9e4293a

Track QueryIterator in num-open-iterators metric

dc0e1c0

add @InterfaceAudience.Public

38a0090

Assert serialized range bounds and extract forwardedRawRangeQuery helper

a913a89

Strengthen and refactor IQv2 headers range integration test

dcc29c9

Jess668 force-pushed the kip-1356-timestamped-range-with-headers-query branch from 1cef606 to dcc29c9 Compare July 9, 2026 14:32

Jess668 marked this pull request as ready for review July 9, 2026 14:45

aliehsaeedii reviewed Jul 9, 2026

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

KAFKA-20704: IQv2 TimestampedRangeWithHeadersQuery for headers-aware key-value stores (KIP-1356)#22770

KAFKA-20704: IQv2 TimestampedRangeWithHeadersQuery for headers-aware key-value stores (KIP-1356)#22770
Jess668 wants to merge 12 commits into
apache:trunkfrom
Jess668:kip-1356-timestamped-range-with-headers-query

Jess668 commented Jul 6, 2026 •

edited

Loading

Uh oh!

Uh oh!

aliehsaeedii left a comment

Uh oh!

aliehsaeedii Jul 9, 2026

Uh oh!

aliehsaeedii Jul 9, 2026

Uh oh!

aliehsaeedii Jul 9, 2026

Uh oh!

aliehsaeedii Jul 9, 2026

Uh oh!

aliehsaeedii Jul 9, 2026

Uh oh!

aliehsaeedii Jul 9, 2026

Uh oh!

aliehsaeedii Jul 9, 2026

Uh oh!

aliehsaeedii Jul 9, 2026

Uh oh!

aliehsaeedii Jul 9, 2026

Uh oh!

aliehsaeedii Jul 9, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

Conversation

Jess668 commented Jul 6, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Dependencies

Summary

Testing

Uh oh!

Uh oh!

aliehsaeedii left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Jess668 commented Jul 6, 2026 •

edited

Loading