mjsax commented on code in PR #14570:
URL: https://github.com/apache/kafka/pull/14570#discussion_r1374842688
##########
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:
As discussed in person: for `RangeQuery`,
`KeyValueToTimestampedKeyValueIteratorAdapter` need to wrap the provided
`RocksDBRangeIterator` to translate between plain-byte[] format what we receive
from the inner store, to the required timestamped-byte[] format that the upper
layer expect.
--
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
