gabriellefu commented on code in PR #23233:
URL: https://github.com/apache/kafka/pull/23233#discussion_r3847918722
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java:
##########
@@ -59,26 +60,17 @@ 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.
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) {
Review Comment:
if there's a adaptor, meaning it's plain inside, just return the plain value
--
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]
