hanyuzheng7 commented on code in PR #14570:
URL: https://github.com/apache/kafka/pull/14570#discussion_r1373970782
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/StoreQueryUtils.java:
##########
@@ -346,6 +346,23 @@ public static <V> Function<byte[], V>
getDeserializeValue(final StateSerdes<?, V
return byteArray -> deserializer.deserialize(serdes.topic(),
byteArray);
}
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ public static <V> Function<byte[], V> getDeserializeValue2(final
StateSerdes<?, V> serdes,
+ final
StateStore wrapped,
+ final boolean
isDSLStore ) {
+ final Serde<V> valueSerde = serdes.valueSerde();
+ final boolean timestamped = WrappedStateStore.isTimestamped(wrapped)
|| isDSLStore;
+ final Deserializer<V> deserializer;
+ if (!timestamped && valueSerde instanceof ValueAndTimestampSerde) {
Review Comment:
Yes, I find that bug, that bug in `WrappedStateStore`, we should add a
statement like this
```
public static boolean isTimestamped(final StateStore stateStore) {
if (stateStore instanceof
KeyValueToTimestampedKeyValueByteStoreAdapter) {
return true;
}
```
But this just solve the ROCK_KV DSL store with cache = true.
Because we the cache = true, we get the data from `CachingKeyValueStore `
but not `KeyValueToTimestampedKeyValueByteStoreAdapter`
But we have to force` timestamped` to be true in the `getDeserializeValue`,
because like `CachingKeyValueStore ` it's not ts-kv store, but ROCK_KV DSL
get data from this store.
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
