Lucy Liu created KAFKA-20798:
--------------------------------
Summary: Add readOnly(IsolationLevel) to
TimestampedWindowStoreWithHeaders and SessionStoreWithHeaders
Key: KAFKA-20798
URL: https://issues.apache.org/jira/browse/KAFKA-20798
Project: Kafka
Issue Type: Bug
Components: streams
Reporter: Lucy Liu
Similar to KAFKA-20497, MeteredTimestampedWindowStoreWithHeaders and
MeteredSessionStoreWithHeaders never override readOnly(IsolationLevel).
KAFKA-20497 added this override to `MeteredKeyValueStore`,
`MeteredWindowStore`, and `MeteredSessionStore` (PRs
[#22316|https://github.com/apache/kafka/pull/22316]/[#22318|https://github.com/apache/kafka/pull/22318]/[#22317|https://github.com/apache/kafka/pull/22317]),
but never extended it to their WithHeaders subclasses, which were added
separately.
Since Kafka Streams' Interactive Query path always calls `.readOnly(level)` on
a store before querying it
(`CompositeReadOnlyWindowStore`/`CompositeReadOnlySessionStore` unconditionally
wrap every underlying store this way), both WithHeaders classes silently fall
back to their superclass's generic ReadOnlyView, which deserializes the key
before the value. That ordering is wrong for headers-aware stores specifically:
the key's schema-ID header can only be recovered from the value's embedded
headers (`ValueTimestampHeaders`/`AggregationWithHeaders`), so the key must be
deserialized after the value, not before.
The result is that any IQ read against a headers-aware windowed or session
store — most reliably reproduced after a restart-and-restore-from-changelog. It
throws:
"org.apache.kafka.common.errors.SerializationException: Error deserializing
schema ID
Caused by: java.lang.IllegalArgumentException: Unknown magic byte!"
Fix: add a readOnly(IsolationLevel) override to both classes, returning a
headers-aware read-only view that reuses the existing correct value-then-key
iterator, matching the pattern KAFKA-20497 established for the non-headers
store classes.
{{}}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
