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;
}
if (stateStore instanceof TimestampedBytesStore) {
return true;
} else if (stateStore instanceof WrappedStateStore) {
```
But this just solve the ROCK_KV DSL store with cache = true.
We have to force` timestamped` to be true.
--
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
