This page is meant as a template for writing a KIP. To create a KIP choose Tools->Copy on this page and modify with your content and replace the heading with the next KIP number and a description of your issue. Replace anything in italics with your own description.

Status

Current state: Accepted

Discussion thread: here 

JIRA: KAFKA-20682 

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

Motivation

KIP-1271 added the ability to persist record headers inside Kafka Streams state stores. When a store is built with one of the new WithHeaders suppliers/builders, the record's Headers are stored alongside the value, surfaced internally through the wrapper types ValueTimestampHeaders<V> (timestamped key/value and window stores) and AggregationWithHeaders<AGG> (session stores).

However, that infrastructure deliberately does not expose headers through Interactive Queries (IQv2): there is no query type whose result carries headers, so even against a header-aware store the stored headers cannot be read back. 
Moreover, support for the existing query types against header stores is itself incomplete and depends on how the store was built. Against an adapter-built header store, queries such as KeyQuery, TimestampedKeyQuery, RangeQuery, TimestampedRangeQuery, WindowKeyQuery, and WindowRangeQuery return only value/timestamp and do not expose headers; against a natively-built header store the same queries fail with UNKNOWN_QUERY_TYPE.

The result is an asymmetry: a user can configure a store to store headers, but has no way to read those headers back via IQv2.

This KIP closes that gap by introducing new IQv2 query types whose results carry headers and by making the existing query types behave consistently against header stores regardless of how the store was built (see Proposed Changes).

Public Interfaces

ReadOnlyRecord

The new query types return their results as a new read-only interface, org.apache.kafka.streams.processor.api.ReadOnlyRecord<K, V>, which exposes a record's key(), value(), timestamp(), and headers() and nothing else. The existing PAPI Record<K, V> is made to implement ReadOnlyRecord<K, V>.

package org.apache.kafka.streams.processor.api;

public interface ReadOnlyRecord<K, V> {
    public K key();
    public V value();
    public long timestamp();
    public Headers headers();
}
package org.apache.kafka.streams.processor.api;

public class Record<K, V> implements ReadOnlyRecord<K, V> {
 	// unchanged   
}


Four new @Evolving, final query classes are added to package org.apache.kafka.streams.query. Each parallels an existing query type and uses the no-get accessor convention (key(), timeFrom(), timeTo(), lowerBound(), upperBound()) throughout — following the newer TimestampedKeyQuery/TimestampedRangeQuery style rather than the older getX() accessors on WindowKeyQuery/WindowRangeQuery, so the new types are internally consistent.

For the iterator queries the result is a ReadOnlyRecordIterator — a closeable iterator that yields ReadOnlyRecord elements directly. 

ReadOnlyRecordIterator

package org.apache.kafka.streams.state;

public interface ReadOnlyRecordIterator<K, V> extends Iterator<ReadOnlyRecord<K, V>>, Closeable {
	@Override
	void close();
}

TimestampedKeyWithHeadersQuery

Parallel of TimestampedKeyQuery. 

package org.apache.kafka.streams.query;

@Evolving
public final class TimestampedKeyWithHeadersQuery<K, V> implements Query<ReadOnlyRecord<K, V>> {
 
	public static <K, V> TimestampedKeyWithHeadersQuery<K, V> withKey(final K key);

	// Skip the cache during query evaluation, forwarding the query to the underlying store. 
	public TimestampedKeyWithHeadersQuery<K, V> skipCache();
 
	public K key(); 
	public boolean isSkipCache();
}

 TimestampedRangeWithHeadersQuery
Parallel of TimestampedRangeQuery.

package org.apache.kafka.streams.query;

@Evolving
public final class TimestampedRangeWithHeadersQuery<K, V> implements Query<ReadOnlyRecordIterator<K, V>> {

	public static <K, V> TimestampedRangeWithHeadersQuery<K, V> withRange(final K lower, final K upper);
	public static <K, V> TimestampedRangeWithHeadersQuery<K, V> withUpperBound(final K upper);
	public static <K, V> TimestampedRangeWithHeadersQuery<K, V> withLowerBound(final K lower);
	public static <K, V> TimestampedRangeWithHeadersQuery<K, V> withNoBounds();

	public TimestampedRangeWithHeadersQuery<K, V> withDescendingKeys();
	public TimestampedRangeWithHeadersQuery<K, V> withAscendingKeys();

	public Optional<K> lowerBound();
	public Optional<K> upperBound();
	public ResultOrder resultOrder();
}

TimestampedWindowKeyWithHeadersQuery
Parallel of WindowKeyQuery. 

Like its counterpart WindowKeyQuery, this type overrides toString(); it is omitted from the signature above as it carries no API contract.

package org.apache.kafka.streams.query;

@Evolving
public final class TimestampedWindowKeyWithHeadersQuery<K, V> implements Query<ReadOnlyRecordIterator<Windowed<K>, V>> {

	public static <K, V> TimestampedWindowKeyWithHeadersQuery<K, V> withKeyAndWindowStartRange(final K key, final Instant timeFrom, final Instant timeTo);

	public K key();
	public Optional<Instant> timeFrom();
	public Optional<Instant> timeTo();
}

TimestampedWindowRangeWithHeadersQuery

Parallel of WindowRangeQuery.

Like its counterpart WindowRangeQuery, this type overrides toString(); it is omitted from the signature above as it carries no API contract.

package org.apache.kafka.streams.query;

@Evolving
public final class TimestampedWindowRangeWithHeadersQuery<K, V> implements Query<ReadOnlyRecordIterator<Windowed<K>, V>> {

	public static <K, V> TimestampedWindowRangeWithHeadersQuery<K, V> withWindowStartRange(final Instant timeFrom, final Instant timeTo);
    public static <K, V> TimestampedWindowRangeWithHeadersQuery<K, V> withKey(final K key);

    public Optional<K> key();
	public Optional<Instant> timeFrom();
	public Optional<Instant> timeTo();
}

Proposed Changes

  • New ReadOnlyRecord interface + Record implements it. As above; the only PAPI change is additive.

  • New query classes returning ReadOnlyRecord-based results.

  • Handler wiring + wrapper→ReadOnlyRecord conversion in the metered header stores. Each metered header store handles its new query types by converting the store's internal wrapper into a ReadOnlyRecord (constructed as a Record):

    • MeteredTimestampedKeyValueStoreWithHeaders handles TimestampedKeyWithHeadersQuery and TimestampedRangeWithHeadersQuery: ValueTimestampHeaders<V> vth → new Record<>(key, vth.value(), vth.timestamp(), vth.headers()).
    • MeteredTimestampedWindowStoreWithHeaders handles TimestampedWindowKeyWithHeadersQuery and the withWindowStartRange form of TimestampedWindowRangeWithHeadersQuery: vth → new Record<>(new Windowed<>(key, window), vth.value(), vth.timestamp(), vth.headers()). timestamp() is the stored record event-time; the window start/end live in the Windowed<K> key.
    • MeteredSessionStoreWithHeaders handles the withKey form of TimestampedWindowRangeWithHeadersQuery: AggregationWithHeaders<AGG> awh → new Record<>(new Windowed<>(key, sessionWindow), awh.aggregation(), sessionWindow.end(), awh.headers()). For sessions the record timestamp is the session-window end (which is why KIP-1271 stores it only in the key, not the value), so timestamp() is filled from window.end().

  • Native header bytes stores now implement query(...) (closes a build-path-dependent gap). Today the existing (non-header) query types succeed or fail depending on how the header store was built:

    • On the adapter build path, the metered store wraps a plain/timestamped byte store with a ToHeadersStoreAdapter that forwards basic IQv2 queries to a normal RocksDBStore (which handles them via StoreQueryUtils) — so the queries succeed.
    • On the native build path, the metered store wraps a native header bytes store (RocksDBTimestampedStoreWithHeaders, RocksDBTimestampedWindowStoreWithHeaders, RocksDBTimeOrderedWindowStoreWithHeaders, RocksDBSessionStoreWithHeaders, RocksDBTimeOrderedSessionStoreWithHeaders), each of which overrides query(...) to return UNKNOWN_QUERY_TYPE for every query — so the queries fail.

    This KIP removes that override and makes the native header bytes stores handle the basic IQv2 queries (returning their stored header-format bytes via StoreQueryUtils.handleBasicQueries, with header-aware (de)serialization). As a result, the existing query types (KeyQuery, TimestampedKeyQuery, RangeQuery, TimestampedRangeQuery on the KV store; WindowKeyQuery, WindowRangeQuery.withWindowStartRange on the window store; WindowRangeQuery.withKey on the session store) behave identically on both build paths, returning header-stripped results, with IQv2 PositionBound and position semantics preserved by StoreQueryUtils. No new public classes are introduced for this part.

  • Behaviour against non-headers stores is unchanged. Submitting one of the new WithHeaders query types to a store that was not built with a WithHeaders supplier produces the standard IQv2 "unknown query type" QueryResult failure.
  • Window time-range constraint. As with the existing window queries, the window WithHeaders queries require a closed window start range (both timeFrom and timeTo present). 


Usage example

// Store built with a WithHeaders supplier (KIP-1271):
//     Stores.timestampedKeyValueStoreWithHeadersBuilder(
//         Stores.persistentTimestampedKeyValueStoreWithHeaders("store"), keySerde, valueSerde)

TimestampedKeyWithHeadersQuery<String, Long> query = TimestampedKeyWithHeadersQuery.withKey("key");

StateQueryRequest<ReadOnlyRecord<String, Long>> request = StateQueryRequest.inStore("store").withQuery(query);

StateQueryResult<ReadOnlyRecord<String, Long>> result = kafkaStreams.query(request);

ReadOnlyRecord<String, Long> rec = result.getOnlyPartitionResult().getResult();
Long value = rec.value();
long timestamp = rec.timestamp();
Headers headers = rec.headers(); // now available via IQv2

Compatibility, Deprecation, and Migration Plan

  • Two new interfaces (ReadOnlyRecord and ReadOnlyRecordIterator) and four new @Evolving query classes. No existing query types, result types, serdes, or store byte formats are changed.
  • Record change is additive. Record now implements ReadOnlyRecord (it already had all four accessors); this is source- and binary-compatible, so existing PAPI code is unaffected. All of these packages live in the same kafka-streams module so the placement imposes no new dependency in either case.
  • Behaviour change for native header stores. Existing query types that previously returned UNKNOWN_QUERY_TYPE against natively built header stores now succeed, returning header-stripped results — matching the behaviour that adapter-built header stores already had. This removes a build-path-dependent inconsistency; no caller that worked before changes behaviour.
  • The new query types' behavior depends on how the store was built:
    • WithHeaders builder + a header-aware (native) supplier → queries succeed and return the stored headers.
    • WithHeaders builder + a non-header supplier (the builder wraps it with a *ToHeadersStoreAdapter, or an in-memory marker) → queries still succeed, but headers() is always empty: the underlying store cannot persist headers, so reads come back with a zero-header value (HeadersBytesStore.convertToHeaderFormat prepends a header count of 0).
    • A plain, non-WithHeaders builder → the new query types are unsupported and fail cleanly with UNKNOWN_QUERY_TYPE.

Because the new types are @Evolving, their API may be refined in a later minor release without a formal deprecation cycle if needed.

Test Plan

New cases are added to the IQv2 integration test suite:

  • shouldHandleTimestampedKeyWithHeadersQuery — asserts the returned ReadOnlyRecord carries the value, timestamp, key, and the exact headers written.
  • shouldHandleTimestampedRangeWithHeadersQuery — asserts each ReadOnlyRecord carries the correct headers, including ascending/descending ordering and the bound variants.
  • shouldHandleTimestampedWindowKeyWithHeadersQuery — asserts windowed ReadOnlyRecords carry headers across the requested window start range, with the window in the Windowed<K> key.
  • shouldHandleTimestampedWindowRangeWithHeadersQuery — covers both forms: withWindowStartRange against the window store and withKey against the session store; asserts headers are carried and that the session ReadOnlyRecord.timestamp() equals the session-window end.
  • Build-path parity (native vs adapter). For each header store, run the existing query types against both a natively built and an adapter-built store and assert identical, header-stripped results — locking in the native-path fix so it cannot regress to UNKNOWN_QUERY_TYPE.
  • Empty-headers cases — for each query type, assert a record written with no headers round-trips as a ReadOnlyRecord whose headers() is empty (an empty RecordHeaders, never null).
  • Tombstone cases — a tombstone is a null value and deletes the key, so a tombstoned key is observably identical to an absent key. Assert that a tombstoned (or never-written) key returns a null result for the point query and is omitted from iterator results.
  • A unit test asserting Record is assignable to ReadOnlyRecord and exposes the same key/value/timestamp/ headers.
  • Negative tests asserting the new query types fail cleanly (unknown-query-type failure) against non-headers stores, and that each store rejects the wrong WindowRangeWithHeadersQuery form.
  • Non-header supplier into a WithHeaders builder — assert the new query types still succeed but return empty headers(), even for records written with headers, since the underlying (non-header) store cannot persist them.
  • No labels