gabriellefu commented on code in PR #23233:
URL: https://github.com/apache/kafka/pull/23233#discussion_r3918173046
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java:
##########
@@ -59,26 +61,26 @@ final class StateManagerUtil {
private StateManagerUtil() {}
static RecordConverter converterForStore(final StateStore store) {
- // First check if the top-level store implements HeadersBytesStore or
TimestampedBytesStore
- if (isHeadersAware(store)) {
- if (store instanceof SessionStore) {
- return rawValueToSessionHeadersValue();
- }
- return rawValueToHeadersValue();
- } else if (isTimestamped(store) && !isVersioned(store)) {
- // should not prepend timestamp when restoring records for
versioned store, as
- // timestamp is used separately during put() process for restore
of versioned stores
- return rawValueToTimestampedValue();
- }
-
- // If top-level check didn't find the type, unwrap to find adapters
- // This handles persistent stores that use adapters
+ // Restore bypasses adapters and writes directly into the inner store,
so the converter must
+ // match the inner store's binary format, not the format the adapter
advertises to the outer
+ // store chain. Thus, check for adapters first.
+ //
+ // This loop enumerates the byte-translating adapters and maps each to
its INNER store's
+ // format. Wrappers that merely advertise a format without translating
bytes (e.g. the
+ // in-memory timestamped markers) are intentionally NOT listed here —
they are resolved by
+ // the isTimestamped()/isHeadersAware() fallback below. Note that
+ // WindowToTimestampedWindowByteStoreAdapter deliberately does not
implement
+ // TimestampedBytesStore: marking it without also converting in its
query() would break the
+ // IQ read path.
StateStore current = store;
while (current != null) {
if (current instanceof TimestampedToHeadersStoreAdapter || current
instanceof TimestampedToHeadersWindowStoreAdapter) {
// Adapter wraps a timestamped store, so restore in
timestamped format
return rawValueToTimestampedValue();
- } else if (current instanceof PlainToHeadersStoreAdapter ||
current instanceof PlainToHeadersWindowStoreAdapter) {
+ } else if (current instanceof PlainToHeadersStoreAdapter
+ || current instanceof PlainToHeadersWindowStoreAdapter
+ || current instanceof
KeyValueToTimestampedKeyValueByteStoreAdapter
+ || current instanceof
WindowToTimestampedWindowByteStoreAdapter) {
Review Comment:
thanks for pointing it out, I have added this branch too
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
