mjsax commented on code in PR #14570:
URL: https://github.com/apache/kafka/pull/14570#discussion_r1379524385
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/KeyValueToTimestampedKeyValueByteStoreAdapter.java:
##########
@@ -127,14 +131,31 @@ public <R> QueryResult<R> query(
final QueryConfig config) {
+
final long start = config.isCollectExecutionInfo() ? System.nanoTime()
: -1L;
- final QueryResult<R> result = store.query(query, positionBound,
config);
+ QueryResult<R> result = store.query(query, positionBound, config);
+ final Position position = result.getPosition();
+
+ if (result.isSuccess()) {
+ if (result.getResult() instanceof byte[]) {
+ final byte[] plainValue = (byte[]) result.getResult();
+ final byte[] valueWithTimestamp =
convertToTimestampedFormat(plainValue);
+ result = (QueryResult<R>)
QueryResult.forResult(valueWithTimestamp);
+ }
+ if (result.getResult() instanceof RocksDbIterator) {
+ final WrappedRocksDbIterator wrappedRocksDBRangeIterator = new
WrappedRocksDbIterator((RocksDbIterator) result.getResult());
+ result = (QueryResult<R>)
QueryResult.forResult(wrappedRocksDBRangeIterator);
+ }
Review Comment:
We should chain this with an `else` that would throw an exception (we should
never hit the `else` and it would be a bug -- throwing an exception would
surface the bug clearly).
--
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
