(kafka) branch trunk updated: KAFKA-20760: Fix ClassCastException in KeyValueToTimestampedKeyValueByteStoreAdapter (#22742)

[bbejeck]

This is an automated email from the ASF dual-hosted git repository.

bbejeck pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new 9ffc78db8a8 KAFKA-20760: Fix ClassCastException in 
KeyValueToTimestampedKeyValueByteStoreAdapter (#22742)
9ffc78db8a8 is described below

commit 9ffc78db8a802b90f3392723d3170767b092aba7
Author: Nick Telford <nick.telf...@gmail.com>
AuthorDate: Fri Jul 3 01:47:51 2026 +0100

    KAFKA-20760: Fix ClassCastException in 
KeyValueToTimestampedKeyValueByteStoreAdapter (#22742)
    
    IQv2 range queries throw a `ClassCastException` when the underlying
    state store is not RocksDB (e.g. an in-memory store), because
    `KeyValueToTimestampedKeyValueByteStoreAdapter` unconditionally cast the
    query iterator result to `RocksDbIterator` before wrapping it. With a
    transactional store the result is a `StagedMergeIterator`, so the cast
    fails.
    
    The adapter exists only to convert plain key/value bytes into the
    timestamped format for any key-value store, so coupling its inner
    iterator to a specific store implementation is unnecessary. Widening the
    field and constructor parameter to `ManagedKeyValueIterator<Bytes,
    byte[]>` — the interface every store iterator already implements —
    removes the erroneous cast and makes the adapter work regardless of the
    underlying store type.
    
    Part of a series of fixes for interactive-query position consistency
    with transactional state stores.
    
    Reviewers: Bill Bejeck <bbej...@apache.org>
    
    Co-authored-by: Bill Bejeck <bbej...@apache.org>
---
 ...eyValueToTimestampedKeyValueByteStoreAdapter.java | 20 ++++++++++----------
 1 file changed, 10 insertions(+), 10 deletions(-)

diff --git 
a/streams/src/main/java/org/apache/kafka/streams/state/internals/KeyValueToTimestampedKeyValueByteStoreAdapter.java
 
b/streams/src/main/java/org/apache/kafka/streams/state/internals/KeyValueToTimestampedKeyValueByteStoreAdapter.java
index 5c5ededb8b4..406823a026a 100644
--- 
a/streams/src/main/java/org/apache/kafka/streams/state/internals/KeyValueToTimestampedKeyValueByteStoreAdapter.java
+++ 
b/streams/src/main/java/org/apache/kafka/streams/state/internals/KeyValueToTimestampedKeyValueByteStoreAdapter.java
@@ -156,8 +156,8 @@ public class KeyValueToTimestampedKeyValueByteStoreAdapter 
implements KeyValueSt
                 final byte[] valueWithTimestamp = 
convertToTimestampedFormat(plainValue);
                 result = (QueryResult<R>) 
InternalQueryResultUtil.copyAndSubstituteDeserializedResult(result, 
valueWithTimestamp);
             } else if (query instanceof RangeQuery || query instanceof 
TimestampedRangeQuery) {
-                final KeyValueToTimestampedKeyValueAdapterIterator 
wrappedRocksDBRangeIterator = new 
KeyValueToTimestampedKeyValueAdapterIterator((RocksDbIterator) 
result.getResult());
-                result = (QueryResult<R>) 
InternalQueryResultUtil.copyAndSubstituteDeserializedResult(result, 
wrappedRocksDBRangeIterator);
+                final KeyValueToTimestampedKeyValueAdapterIterator 
wrappedIterator = new 
KeyValueToTimestampedKeyValueAdapterIterator((ManagedKeyValueIterator<Bytes, 
byte[]>) result.getResult());
+                result = (QueryResult<R>) 
InternalQueryResultUtil.copyAndSubstituteDeserializedResult(result, 
wrappedIterator);
             } else {
                 throw new IllegalArgumentException("Unsupported query type: " 
+ query.getClass());
             }
@@ -218,35 +218,35 @@ public class 
KeyValueToTimestampedKeyValueByteStoreAdapter implements KeyValueSt
 
     private static class KeyValueToTimestampedKeyValueAdapterIterator 
implements ManagedKeyValueIterator<Bytes, byte[]> {
 
-        private final RocksDbIterator rocksDbIterator;
+        private final ManagedKeyValueIterator<Bytes, byte[]> iterator;
 
-        public KeyValueToTimestampedKeyValueAdapterIterator(final 
RocksDbIterator rocksDbIterator) {
-            this.rocksDbIterator = rocksDbIterator;
+        public KeyValueToTimestampedKeyValueAdapterIterator(final 
ManagedKeyValueIterator<Bytes, byte[]> iterator) {
+            this.iterator = iterator;
         }
 
         @Override
         public void close() {
-            rocksDbIterator.close();
+            iterator.close();
         }
 
         @Override
         public Bytes peekNextKey() {
-            return rocksDbIterator.peekNextKey();
+            return iterator.peekNextKey();
         }
 
         @Override
         public void onClose(final Runnable closeCallback) {
-            rocksDbIterator.onClose(closeCallback);
+            iterator.onClose(closeCallback);
         }
 
         @Override
         public boolean hasNext() {
-            return rocksDbIterator.hasNext();
+            return iterator.hasNext();
         }
 
         @Override
         public KeyValue<Bytes, byte[]> next() {
-            final KeyValue<Bytes, byte[]> next = rocksDbIterator.next();
+            final KeyValue<Bytes, byte[]> next = iterator.next();
             if (next == null) {
                 return null;
             }