[
https://issues.apache.org/jira/browse/KAFKA-21024?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Ziyun Fu updated KAFKA-21024:
-----------------------------
Description:
Follow-up from KAFKA-16158
KAFKA-16158 fixed KeyValueToTimestampedKeyValueByteStoreAdapter to implement
TimestampedBytesStore, since every one of its read paths (get/range/all/query)
already converted the wrapped plain store's bytes to timestamped format via
TimestampedBytesStore#convertToTimestampedFormat.
WindowToTimestampedWindowByteStoreAdapter has the same kind of wrapping
responsibility but is inconsistent between its two read paths:
- fetch() convert the wrapped plain store's bytes to timestamped format
- query() delegates directly to the wrapped store with no conversion at all,
so callers going through WindowKeyQuery / WindowRangeQuery get back
plain-format bytes.
Because of this gap, KAFKA-16158 deliberately did NOT mark
WindowToTimestampedWindowByteStoreAdapter as TimestampedBytesStore -- doing so
would make marker-based callers assume the query() path also returns
timestamped bytes, which would break the IQ read path for window stores. See
the comment above the adapter-enumeration loop in
StateManagerUtil#converterForStore.
Tasks:
# Update WindowToTimestampedWindowByteStoreAdapter#query() to convert results
to timestamped format for WindowKeyQuery / WindowRangeQuery (and any other
applicable IQv2 query types), mirroring what
KeyValueToTimestampedKeyValueByteStoreAdapter#query() already does for KeyQuery
/ TimestampedKeyQuery / RangeQuery.
# Once query() is consistent with fetch(), have
WindowToTimestampedWindowByteStoreAdapter implement TimestampedBytesStore.
# Update/remove the now-stale comment in StateManagerUtil#converterForStore
that explains why the window adapter is excluded from the marker, and revisit
whether it still needs to be listed explicitly in the adapter-enumeration
branch (KeyValueToTimestampedKeyValueByteStoreAdapter still needs to be there
for restore purposes even after implementing the marker, since restore bypasses
the adapter and must match the inner store's raw format).
was:
{panel}
h3. Context & Problem Statement
*KAFKA-16158* updated {{KeyValueToTimestampedKeyValueByteStoreAdapter}} to
implement {{{}TimestampedBytesStore{}}}, ensuring all read paths ({{{}get{}}},
{{{}range{}}}, {{{}all{}}}, {{{}query{}}}) convert the wrapped store's bytes
via {{{}TimestampedBytesStore#convertToTimestampedFormat{}}}.
{{WindowToTimestampedWindowByteStoreAdapter}} serves a similar wrapper role but
handles its two read paths inconsistently:
*
*{{fetch()}} (and iterator overloads):* Properly converts the wrapped plain
store's bytes to timestamped format (via
{{WindowToTimestampedWindowByteStoreAdapter#fetch(Bytes, long)}} and
{{{}WindowToTimestampedWindowIteratorAdapter{}}}).
*
*{{query()}} (IQv2 path):* Delegates directly to the wrapped store without
conversion, returning raw/plain-format bytes for {{WindowKeyQuery}} and
{{{}WindowRangeQuery{}}}.
Due to this inconsistency, {{WindowToTimestampedWindowByteStoreAdapter}} was
intentionally *not* marked as {{{}TimestampedBytesStore{}}}. Doing so would
cause marker-based callers (e.g., {{{}StateManagerUtil#converterForStore{}}},
{{{}WrappedStateStore#isTimestamped{}}}) to wrongly assume {{query()}} returned
timestamped bytes, breaking IQ read paths for window stores.
h3. Implementation Tasks
#
*Update {{WindowToTimestampedWindowByteStoreAdapter#query()}}*
**
Convert query results to timestamped format for {{{}WindowKeyQuery{}}},
{{{}WindowRangeQuery{}}}, and any other applicable IQv2 query types.
**
Align behavior with {{KeyValueToTimestampedKeyValueByteStoreAdapter#query()}}
(which handles {{{}KeyQuery{}}}, {{{}TimestampedKeyQuery{}}}, and
{{{}RangeQuery{}}}).
#
*Implement {{TimestampedBytesStore}}*
**
Once {{query()}} matches {{fetch()}} in returning timestamped bytes, update
{{WindowToTimestampedWindowByteStoreAdapter}} to implement
{{{}TimestampedBytesStore{}}}.
#
*Clean up {{StateManagerUtil#converterForStore}}*
**
Update or remove the stale comment explaining why the window adapter was
excluded from the marker interface.
**
Revisit whether {{WindowToTimestampedWindowByteStoreAdapter}} needs explicit
listing in the adapter-enumeration branch.
**
_Note:_ {{KeyValueToTimestampedKeyValueByteStoreAdapter}} must remain
explicitly enumerated for restore operations because restore bypasses the
adapter and requires matching the inner store's raw byte format.{panel}
> WindowToTimestampedWindowByteStoreAdapter.query() does not convert results to
> timestamped format
> ------------------------------------------------------------------------------------------------------------
>
> Key: KAFKA-21024
> URL: https://issues.apache.org/jira/browse/KAFKA-21024
> Project: Kafka
> Issue Type: Improvement
> Reporter: Ziyun Fu
> Priority: Major
>
> Follow-up from KAFKA-16158
> KAFKA-16158 fixed KeyValueToTimestampedKeyValueByteStoreAdapter to implement
> TimestampedBytesStore, since every one of its read paths
> (get/range/all/query) already converted the wrapped plain store's bytes to
> timestamped format via TimestampedBytesStore#convertToTimestampedFormat.
> WindowToTimestampedWindowByteStoreAdapter has the same kind of wrapping
> responsibility but is inconsistent between its two read paths:
> - fetch() convert the wrapped plain store's bytes to timestamped format
> - query() delegates directly to the wrapped store with no conversion at
> all, so callers going through WindowKeyQuery / WindowRangeQuery get back
> plain-format bytes.
> Because of this gap, KAFKA-16158 deliberately did NOT mark
> WindowToTimestampedWindowByteStoreAdapter as TimestampedBytesStore -- doing
> so would make marker-based callers assume the query() path also returns
> timestamped bytes, which would break the IQ read path for window stores. See
> the comment above the adapter-enumeration loop in
> StateManagerUtil#converterForStore.
> Tasks:
> # Update WindowToTimestampedWindowByteStoreAdapter#query() to convert
> results to timestamped format for WindowKeyQuery / WindowRangeQuery (and any
> other applicable IQv2 query types), mirroring what
> KeyValueToTimestampedKeyValueByteStoreAdapter#query() already does for
> KeyQuery / TimestampedKeyQuery / RangeQuery.
> # Once query() is consistent with fetch(), have
> WindowToTimestampedWindowByteStoreAdapter implement TimestampedBytesStore.
> # Update/remove the now-stale comment in StateManagerUtil#converterForStore
> that explains why the window adapter is excluded from the marker, and revisit
> whether it still needs to be listed explicitly in the adapter-enumeration
> branch (KeyValueToTimestampedKeyValueByteStoreAdapter still needs to be there
> for restore purposes even after implementing the marker, since restore
> bypasses the adapter and must match the inner store's raw format).
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
