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 04bfe7dbb0f KAFKA-20497: Add TimeOrderedCachingWindowStore#readOnly
(#22315)
04bfe7dbb0f is described below
commit 04bfe7dbb0f8760022afe51647a2e7dd7f434288
Author: Nick Telford <[email protected]>
AuthorDate: Mon Jun 15 18:10:35 2026 +0100
KAFKA-20497: Add TimeOrderedCachingWindowStore#readOnly (#22315)
The cache holds uncommitted writes that must not be visible under
READ_COMMITTED, so that isolation level bypasses the cache entirely and
delegates straight to the inner store's readOnly view. READ_UNCOMMITTED
requires a merged view through a ReadOnlyView. The ReadOnlyView
delegates to the existing fetchInternal/fetchKeyRange/fetchAllInternal
helpers and converts Instant arguments to epoch-milliseconds before
calling them; the cache key schema operates in longs throughout, so
converting early avoids repeated Instant-to-long conversions in iterator
hot paths.
KAFKA-20497
Reviewers: Bill Bejeck <[email protected]>
---
.../internals/TimeOrderedCachingWindowStore.java | 234 +++++++++++++--------
...imeOrderedCachingPersistentWindowStoreTest.java | 175 ++++++++++++++-
.../internals/TimeOrderedWindowStoreTest.java | 3 +-
3 files changed, 328 insertions(+), 84 deletions(-)
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/TimeOrderedCachingWindowStore.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/TimeOrderedCachingWindowStore.java
index 4e0fd29f09d..b11920c8536 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/TimeOrderedCachingWindowStore.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/TimeOrderedCachingWindowStore.java
@@ -16,12 +16,14 @@
*/
package org.apache.kafka.streams.state.internals;
+import org.apache.kafka.common.IsolationLevel;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.internals.ApiUtils;
import org.apache.kafka.streams.kstream.Windowed;
import org.apache.kafka.streams.kstream.internals.Change;
import org.apache.kafka.streams.processor.StateStore;
@@ -32,6 +34,7 @@ import
org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
import org.apache.kafka.streams.processor.internals.ProcessorStateManager;
import org.apache.kafka.streams.processor.internals.RecordQueue;
import org.apache.kafka.streams.state.KeyValueIterator;
+import org.apache.kafka.streams.state.ReadOnlyWindowStore;
import org.apache.kafka.streams.state.StateSerdes;
import org.apache.kafka.streams.state.WindowStore;
import org.apache.kafka.streams.state.WindowStoreIterator;
@@ -45,11 +48,13 @@ import
org.apache.kafka.streams.state.internals.ThreadCache.DirtyEntry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.time.Instant;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
+import java.util.Objects;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Function;
@@ -302,59 +307,48 @@ public class TimeOrderedCachingWindowStore
@Override
public byte[] fetch(final Bytes key,
final long timestamp) {
+ return fetchPointInternal(wrapped(), key, timestamp);
+ }
+
+ private byte[] fetchPointInternal(final ReadOnlyWindowStore<Bytes, byte[]>
store,
+ final Bytes key,
+ final long timestamp) {
validateStoreOpen();
if (internalContext.cache() == null) {
- return wrapped().fetch(key, timestamp);
+ return store.fetch(key, timestamp);
}
-
final Bytes baseBytesKey =
TimeFirstWindowKeySchema.toStoreKeyBinary(key, timestamp, 0);
final Bytes cacheKey = baseKeyCacheFunction.cacheKey(baseBytesKey);
-
final LRUCacheEntry entry = internalContext.cache().get(cacheName,
cacheKey);
- if (entry == null) {
- return wrapped().fetch(key, timestamp);
- } else {
- return entry.value();
- }
+ return entry == null ? store.fetch(key, timestamp) : entry.value();
}
@Override
- public synchronized WindowStoreIterator<byte[]> fetch(final Bytes key,
- final long timeFrom,
- final long timeTo) {
- // since this function may not access the underlying inner store, we
need to validate
- // if store is open outside as well.
- validateStoreOpen();
-
- final WindowStoreIterator<byte[]> underlyingIterator =
wrapped().fetch(key, timeFrom, timeTo);
- if (internalContext.cache() == null) {
- return underlyingIterator;
- }
-
- return fetchInternal(underlyingIterator, key, timeFrom, timeTo, true);
+ public WindowStoreIterator<byte[]> fetch(final Bytes key,
+ final long timeFrom,
+ final long timeTo) {
+ return fetchInternal(wrapped(), key, timeFrom, timeTo, true);
}
@Override
- public synchronized WindowStoreIterator<byte[]> backwardFetch(final Bytes
key,
- final long
timeFrom,
- final long
timeTo) {
- // since this function may not access the underlying inner store, we
need to validate
- // if store is open outside as well.
- validateStoreOpen();
+ public WindowStoreIterator<byte[]> backwardFetch(final Bytes key,
+ final long timeFrom,
+ final long timeTo) {
+ return fetchInternal(wrapped(), key, timeFrom, timeTo, false);
+ }
- final WindowStoreIterator<byte[]> underlyingIterator =
wrapped().backwardFetch(key, timeFrom, timeTo);
+ private synchronized WindowStoreIterator<byte[]> fetchInternal(final
ReadOnlyWindowStore<Bytes, byte[]> store,
+ final Bytes
key,
+ final long
timeFrom,
+ final long
timeTo,
+ final
boolean forward) {
+ validateStoreOpen();
+ final WindowStoreIterator<byte[]> underlyingIterator = forward
+ ? store.fetch(key, Instant.ofEpochMilli(timeFrom),
Instant.ofEpochMilli(timeTo))
+ : store.backwardFetch(key, Instant.ofEpochMilli(timeFrom),
Instant.ofEpochMilli(timeTo));
if (internalContext.cache() == null) {
return underlyingIterator;
}
-
- return fetchInternal(underlyingIterator, key, timeFrom, timeTo, false);
- }
-
- private WindowStoreIterator<byte[]> fetchInternal(final
WindowStoreIterator<byte[]> underlyingIterator,
- final Bytes key,
- final long timeFrom,
- final long timeTo,
- final boolean forward) {
final PeekingKeyValueIterator<Bytes, LRUCacheEntry> cacheIterator =
new CacheIteratorWrapper(
key, timeFrom, timeTo, forward, hasIndex);
final KeySchema keySchema = hasIndex ? indexKeySchema : baseKeySchema;
@@ -374,25 +368,7 @@ public class TimeOrderedCachingWindowStore
final Bytes keyTo,
final long timeFrom,
final long timeTo) {
- if (keyFrom != null && keyTo != null && keyFrom.compareTo(keyTo) > 0) {
- LOG.warn("Returning empty iterator for fetch with invalid key
range: from > to. " +
- "This may be due to range arguments set in the wrong order, " +
- "or serdes that don't preserve ordering when lexicographically
comparing the serialized bytes. " +
- "Note that the built-in numerical serdes do not follow this
for negative numbers");
- return KeyValueIterators.emptyIterator();
- }
-
- // since this function may not access the underlying inner store, we
need to validate
- // if store is open outside as well.
- validateStoreOpen();
-
- final KeyValueIterator<Windowed<Bytes>, byte[]> underlyingIterator =
- wrapped().fetch(keyFrom, keyTo, timeFrom, timeTo);
- if (internalContext.cache() == null) {
- return underlyingIterator;
- }
-
- return fetchKeyRange(underlyingIterator, keyFrom, keyTo, timeFrom,
timeTo, true);
+ return fetchKeyRange(wrapped(), keyFrom, keyTo, timeFrom, timeTo,
true);
}
@Override
@@ -400,32 +376,25 @@ public class TimeOrderedCachingWindowStore
final Bytes
keyTo,
final long
timeFrom,
final long
timeTo) {
- if (keyFrom != null && keyTo != null && keyFrom.compareTo(keyTo) > 0) {
- LOG.warn("Returning empty iterator for fetch with invalid key
range: from > to. "
- + "This may be due to serdes that don't preserve ordering when
lexicographically comparing the serialized bytes. " +
- "Note that the built-in numerical serdes do not follow this
for negative numbers");
- return KeyValueIterators.emptyIterator();
- }
-
- // since this function may not access the underlying inner store, we
need to validate
- // if store is open outside as well.
- validateStoreOpen();
-
- final KeyValueIterator<Windowed<Bytes>, byte[]> underlyingIterator =
- wrapped().backwardFetch(keyFrom, keyTo, timeFrom, timeTo);
- if (internalContext.cache() == null) {
- return underlyingIterator;
- }
-
- return fetchKeyRange(underlyingIterator, keyFrom, keyTo, timeFrom,
timeTo, false);
+ return fetchKeyRange(wrapped(), keyFrom, keyTo, timeFrom, timeTo,
false);
}
- private KeyValueIterator<Windowed<Bytes>, byte[]> fetchKeyRange(final
KeyValueIterator<Windowed<Bytes>, byte[]> underlyingIterator,
+ private KeyValueIterator<Windowed<Bytes>, byte[]> fetchKeyRange(final
ReadOnlyWindowStore<Bytes, byte[]> store,
final
Bytes keyFrom,
final
Bytes keyTo,
final long
timeFrom,
final long
timeTo,
final
boolean forward) {
+ if (isInvalidKeyRange(keyFrom, keyTo)) {
+ return KeyValueIterators.emptyIterator();
+ }
+ validateStoreOpen();
+ final KeyValueIterator<Windowed<Bytes>, byte[]> underlyingIterator =
forward
+ ? store.fetch(keyFrom, keyTo, Instant.ofEpochMilli(timeFrom),
Instant.ofEpochMilli(timeTo))
+ : store.backwardFetch(keyFrom, keyTo,
Instant.ofEpochMilli(timeFrom), Instant.ofEpochMilli(timeTo));
+ if (internalContext.cache() == null) {
+ return underlyingIterator;
+ }
final PeekingKeyValueIterator<Bytes, LRUCacheEntry> cacheIterator =
new CacheIteratorWrapper(
keyFrom, keyTo, timeFrom, timeTo, forward, hasIndex);
@@ -453,19 +422,24 @@ public class TimeOrderedCachingWindowStore
@Override
public KeyValueIterator<Windowed<Bytes>, byte[]> fetchAll(final long
timeFrom,
final long
timeTo) {
- validateStoreOpen();
-
- final KeyValueIterator<Windowed<Bytes>, byte[]> underlyingIterator =
wrapped().fetchAll(timeFrom, timeTo);
- return fetchAllInternal(underlyingIterator, timeFrom, timeTo, true);
+ return fetchAllInternal(wrapped(), timeFrom, timeTo, true);
}
@Override
public KeyValueIterator<Windowed<Bytes>, byte[]> backwardFetchAll(final
long timeFrom,
final
long timeTo) {
- validateStoreOpen();
+ return fetchAllInternal(wrapped(), timeFrom, timeTo, false);
+ }
- final KeyValueIterator<Windowed<Bytes>, byte[]> underlyingIterator =
wrapped().backwardFetchAll(timeFrom, timeTo);
- return fetchAllInternal(underlyingIterator, timeFrom, timeTo, false);
+ private KeyValueIterator<Windowed<Bytes>, byte[]> fetchAllInternal(final
ReadOnlyWindowStore<Bytes, byte[]> store,
+ final
long timeFrom,
+ final
long timeTo,
+ final
boolean forward) {
+ validateStoreOpen();
+ final KeyValueIterator<Windowed<Bytes>, byte[]> underlyingIterator =
forward
+ ? store.fetchAll(Instant.ofEpochMilli(timeFrom),
Instant.ofEpochMilli(timeTo))
+ : store.backwardFetchAll(Instant.ofEpochMilli(timeFrom),
Instant.ofEpochMilli(timeTo));
+ return fetchAllInternal(underlyingIterator, timeFrom, timeTo, forward);
}
private KeyValueIterator<Windowed<Bytes>, byte[]> fetchAllInternal(final
KeyValueIterator<Windowed<Bytes>, byte[]> underlyingIterator,
@@ -538,6 +512,102 @@ public class TimeOrderedCachingWindowStore
}
}
+ @Override
+ public ReadOnlyWindowStore<Bytes, byte[]> readOnly(final IsolationLevel
isolationLevel) {
+ Objects.requireNonNull(isolationLevel, "isolationLevel cannot be
null");
+ if (isolationLevel == IsolationLevel.READ_COMMITTED) {
+ return wrapped().readOnly(isolationLevel);
+ }
+ return new ReadOnlyView(wrapped().readOnly(isolationLevel));
+ }
+
+ private boolean isInvalidKeyRange(final Bytes keyFrom, final Bytes keyTo) {
+ if (keyFrom != null && keyTo != null && keyFrom.compareTo(keyTo) > 0) {
+ LOG.warn("Returning empty iterator for fetch with invalid key
range: from > to. " +
+ "This may be due to range arguments set in the wrong
order, " +
+ "or serdes that don't preserve ordering when
lexicographically comparing the serialized bytes. " +
+ "Note that the built-in numerical serdes do not follow
this for negative numbers");
+ return true;
+ }
+ return false;
+ }
+
+ private final class ReadOnlyView implements ReadOnlyWindowStore<Bytes,
byte[]> {
+
+ private final ReadOnlyWindowStore<Bytes, byte[]> underlying;
+
+ ReadOnlyView(final ReadOnlyWindowStore<Bytes, byte[]> underlying) {
+ this.underlying = underlying;
+ }
+
+ @Override
+ public byte[] fetch(final Bytes key, final long timestamp) {
+ return fetchPointInternal(underlying, key, timestamp);
+ }
+
+ @Override
+ public WindowStoreIterator<byte[]> fetch(final Bytes key, final
Instant timeFrom, final Instant timeTo) {
+ return fetchInternal(underlying, key,
+ ApiUtils.validateMillisecondInstant(timeFrom,
ApiUtils.prepareMillisCheckFailMsgPrefix(timeFrom, "timeFrom")),
+ ApiUtils.validateMillisecondInstant(timeTo,
ApiUtils.prepareMillisCheckFailMsgPrefix(timeTo, "timeTo")),
+ true);
+ }
+
+ @Override
+ public WindowStoreIterator<byte[]> backwardFetch(final Bytes key,
final Instant timeFrom, final Instant timeTo) {
+ return fetchInternal(underlying, key,
+ ApiUtils.validateMillisecondInstant(timeFrom,
ApiUtils.prepareMillisCheckFailMsgPrefix(timeFrom, "timeFrom")),
+ ApiUtils.validateMillisecondInstant(timeTo,
ApiUtils.prepareMillisCheckFailMsgPrefix(timeTo, "timeTo")),
+ false);
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]> fetch(final Bytes
keyFrom, final Bytes keyTo, final Instant timeFrom, final Instant timeTo) {
+ return fetchKeyRange(underlying, keyFrom, keyTo,
+ ApiUtils.validateMillisecondInstant(timeFrom,
ApiUtils.prepareMillisCheckFailMsgPrefix(timeFrom, "timeFrom")),
+ ApiUtils.validateMillisecondInstant(timeTo,
ApiUtils.prepareMillisCheckFailMsgPrefix(timeTo, "timeTo")),
+ true);
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]> backwardFetch(final
Bytes keyFrom, final Bytes keyTo, final Instant timeFrom, final Instant timeTo)
{
+ return fetchKeyRange(underlying, keyFrom, keyTo,
+ ApiUtils.validateMillisecondInstant(timeFrom,
ApiUtils.prepareMillisCheckFailMsgPrefix(timeFrom, "timeFrom")),
+ ApiUtils.validateMillisecondInstant(timeTo,
ApiUtils.prepareMillisCheckFailMsgPrefix(timeTo, "timeTo")),
+ false);
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]> fetchAll(final
Instant timeFrom, final Instant timeTo) {
+ return fetchAllInternal(underlying,
+ ApiUtils.validateMillisecondInstant(timeFrom,
ApiUtils.prepareMillisCheckFailMsgPrefix(timeFrom, "timeFrom")),
+ ApiUtils.validateMillisecondInstant(timeTo,
ApiUtils.prepareMillisCheckFailMsgPrefix(timeTo, "timeTo")),
+ true);
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]>
backwardFetchAll(final Instant timeFrom, final Instant timeTo) {
+ return fetchAllInternal(underlying,
+ ApiUtils.validateMillisecondInstant(timeFrom,
ApiUtils.prepareMillisCheckFailMsgPrefix(timeFrom, "timeFrom")),
+ ApiUtils.validateMillisecondInstant(timeTo,
ApiUtils.prepareMillisCheckFailMsgPrefix(timeTo, "timeTo")),
+ false);
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]> all() {
+ validateStoreOpen();
+ final KeyValueIterator<Windowed<Bytes>, byte[]> underlyingIterator
= underlying.all();
+ return fetchAllInternal(underlyingIterator, 0, Long.MAX_VALUE,
true);
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]> backwardAll() {
+ validateStoreOpen();
+ final KeyValueIterator<Windowed<Bytes>, byte[]> underlyingIterator
= underlying.backwardAll();
+ return fetchAllInternal(underlyingIterator, 0, Long.MAX_VALUE,
false);
+ }
+ }
+
private class CacheIteratorWrapper implements
PeekingKeyValueIterator<Bytes, LRUCacheEntry> {
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/internals/TimeOrderedCachingPersistentWindowStoreTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/TimeOrderedCachingPersistentWindowStoreTest.java
index b1148fa0716..2245e1eab00 100644
---
a/streams/src/test/java/org/apache/kafka/streams/state/internals/TimeOrderedCachingPersistentWindowStoreTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/state/internals/TimeOrderedCachingPersistentWindowStoreTest.java
@@ -17,6 +17,7 @@
package org.apache.kafka.streams.state.internals;
import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.IsolationLevel;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.serialization.IntegerSerializer;
@@ -43,6 +44,7 @@ import
org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
import org.apache.kafka.streams.query.Position;
import org.apache.kafka.streams.state.KeyValueIterator;
+import org.apache.kafka.streams.state.ReadOnlyWindowStore;
import org.apache.kafka.streams.state.StoreBuilder;
import org.apache.kafka.streams.state.Stores;
import org.apache.kafka.streams.state.TimestampedWindowStore;
@@ -1219,7 +1221,8 @@ public class TimeOrderedCachingPersistentWindowStoreTest {
assertThat(
messages,
hasItem("Returning empty iterator for fetch with invalid key
range: from > to." +
- " This may be due to serdes that don't preserve ordering
when lexicographically comparing the serialized bytes." +
+ " This may be due to range arguments set in the wrong
order, " +
+ "or serdes that don't preserve ordering when
lexicographically comparing the serialized bytes." +
" Note that the built-in numerical serdes do not follow
this for negative numbers")
);
}
@@ -1256,6 +1259,176 @@ public class
TimeOrderedCachingPersistentWindowStoreTest {
verifyAndTearDownCloseTests();
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void shouldReadCommittedBypassesCache(final boolean hasIndex) {
+ setUp(hasIndex);
+ cachingStore.put(bytesKey("a"), bytesValue("a"), DEFAULT_TIMESTAMP);
+
+ try (final WindowStoreIterator<byte[]> it =
+ cachingStore.readOnly(IsolationLevel.READ_COMMITTED)
+ .fetch(bytesKey("a"), ofEpochMilli(DEFAULT_TIMESTAMP),
ofEpochMilli(DEFAULT_TIMESTAMP))) {
+ assertFalse(it.hasNext());
+ }
+
+ cachingStore.commit(Map.of());
+
+ try (final WindowStoreIterator<byte[]> it =
+ cachingStore.readOnly(IsolationLevel.READ_COMMITTED)
+ .fetch(bytesKey("a"), ofEpochMilli(DEFAULT_TIMESTAMP),
ofEpochMilli(DEFAULT_TIMESTAMP))) {
+ assertTrue(it.hasNext());
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void shouldReadUncommittedViewFetchPointInTime(final boolean
hasIndex) {
+ setUp(hasIndex);
+ cachingStore.put(bytesKey("a"), bytesValue("store"),
DEFAULT_TIMESTAMP);
+ cachingStore.commit(Map.of());
+ cachingStore.put(bytesKey("b"), bytesValue("cache"),
DEFAULT_TIMESTAMP);
+
+ final ReadOnlyWindowStore<Bytes, byte[]> view =
cachingStore.readOnly(IsolationLevel.READ_UNCOMMITTED);
+ assertArrayEquals(bytesValue("store"), view.fetch(bytesKey("a"),
DEFAULT_TIMESTAMP));
+ assertArrayEquals(bytesValue("cache"), view.fetch(bytesKey("b"),
DEFAULT_TIMESTAMP));
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void
shouldReadUncommittedViewFetchSingleKeyMergesCacheAndStore(final boolean
hasIndex) {
+ setUp(hasIndex);
+ cachingStore.put(bytesKey("a"), bytesValue("1"), DEFAULT_TIMESTAMP);
+ cachingStore.commit(Map.of());
+ cachingStore.put(bytesKey("a"), bytesValue("2"), DEFAULT_TIMESTAMP +
20);
+
+ final ReadOnlyWindowStore<Bytes, byte[]> view =
cachingStore.readOnly(IsolationLevel.READ_UNCOMMITTED);
+ try (final WindowStoreIterator<byte[]> it =
+ view.fetch(bytesKey("a"), ofEpochMilli(DEFAULT_TIMESTAMP),
ofEpochMilli(DEFAULT_TIMESTAMP + 20))) {
+ verifyKeyValue(it.next(), DEFAULT_TIMESTAMP, "1");
+ verifyKeyValue(it.next(), DEFAULT_TIMESTAMP + 20, "2");
+ assertFalse(it.hasNext());
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void
shouldReadUncommittedViewBackwardFetchSingleKeyMergesCacheAndStore(final
boolean hasIndex) {
+ setUp(hasIndex);
+ cachingStore.put(bytesKey("a"), bytesValue("1"), DEFAULT_TIMESTAMP);
+ cachingStore.commit(Map.of());
+ cachingStore.put(bytesKey("a"), bytesValue("2"), DEFAULT_TIMESTAMP +
20);
+
+ final ReadOnlyWindowStore<Bytes, byte[]> view =
cachingStore.readOnly(IsolationLevel.READ_UNCOMMITTED);
+ try (final WindowStoreIterator<byte[]> it =
+ view.backwardFetch(bytesKey("a"),
ofEpochMilli(DEFAULT_TIMESTAMP), ofEpochMilli(DEFAULT_TIMESTAMP + 20))) {
+ verifyKeyValue(it.next(), DEFAULT_TIMESTAMP + 20, "2");
+ verifyKeyValue(it.next(), DEFAULT_TIMESTAMP, "1");
+ assertFalse(it.hasNext());
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void shouldReadUncommittedViewFetchRangeMergesCacheAndStore(final
boolean hasIndex) {
+ setUp(hasIndex);
+ cachingStore.put(bytesKey("a"), bytesValue("1"), DEFAULT_TIMESTAMP);
+ cachingStore.commit(Map.of());
+ cachingStore.put(bytesKey("b"), bytesValue("2"), DEFAULT_TIMESTAMP);
+
+ final ReadOnlyWindowStore<Bytes, byte[]> view =
cachingStore.readOnly(IsolationLevel.READ_UNCOMMITTED);
+ final List<KeyValue<Windowed<Bytes>, byte[]>> results =
toListAndCloseIterator(
+ view.fetch(bytesKey("a"), bytesKey("b"),
ofEpochMilli(DEFAULT_TIMESTAMP), ofEpochMilli(DEFAULT_TIMESTAMP)));
+ assertEquals(2, results.size());
+ assertEquals(bytesKey("a"), results.get(0).key.key());
+ assertEquals(bytesKey("b"), results.get(1).key.key());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void
shouldReadUncommittedViewBackwardFetchRangeMergesCacheAndStore(final boolean
hasIndex) {
+ setUp(hasIndex);
+ cachingStore.put(bytesKey("a"), bytesValue("1"), DEFAULT_TIMESTAMP);
+ cachingStore.commit(Map.of());
+ cachingStore.put(bytesKey("b"), bytesValue("2"), DEFAULT_TIMESTAMP);
+
+ final ReadOnlyWindowStore<Bytes, byte[]> view =
cachingStore.readOnly(IsolationLevel.READ_UNCOMMITTED);
+ final List<KeyValue<Windowed<Bytes>, byte[]>> results =
toListAndCloseIterator(
+ view.backwardFetch(bytesKey("a"), bytesKey("b"),
ofEpochMilli(DEFAULT_TIMESTAMP), ofEpochMilli(DEFAULT_TIMESTAMP)));
+ assertEquals(2, results.size());
+ assertEquals(bytesKey("b"), results.get(0).key.key());
+ assertEquals(bytesKey("a"), results.get(1).key.key());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void shouldReadUncommittedViewFetchAllMergesCacheAndStore(final
boolean hasIndex) {
+ setUp(hasIndex);
+ cachingStore.put(bytesKey("a"), bytesValue("1"), DEFAULT_TIMESTAMP);
+ cachingStore.commit(Map.of());
+ cachingStore.put(bytesKey("b"), bytesValue("2"), DEFAULT_TIMESTAMP);
+
+ final ReadOnlyWindowStore<Bytes, byte[]> view =
cachingStore.readOnly(IsolationLevel.READ_UNCOMMITTED);
+ final List<KeyValue<Windowed<Bytes>, byte[]>> results =
toListAndCloseIterator(
+ view.fetchAll(ofEpochMilli(DEFAULT_TIMESTAMP),
ofEpochMilli(DEFAULT_TIMESTAMP)));
+ assertEquals(2, results.size());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void shouldReadUncommittedViewAllMergesCacheAndStore(final boolean
hasIndex) {
+ setUp(hasIndex);
+ cachingStore.put(bytesKey("a"), bytesValue("1"), DEFAULT_TIMESTAMP);
+ cachingStore.commit(Map.of());
+ cachingStore.put(bytesKey("b"), bytesValue("2"), DEFAULT_TIMESTAMP);
+
+ final ReadOnlyWindowStore<Bytes, byte[]> view =
cachingStore.readOnly(IsolationLevel.READ_UNCOMMITTED);
+ final List<KeyValue<Windowed<Bytes>, byte[]>> results =
toListAndCloseIterator(view.all());
+ assertEquals(2, results.size());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void shouldReadUncommittedViewBackwardAllMergesCacheAndStore(final
boolean hasIndex) {
+ setUp(hasIndex);
+ cachingStore.put(bytesKey("a"), bytesValue("1"), DEFAULT_TIMESTAMP);
+ cachingStore.commit(Map.of());
+ cachingStore.put(bytesKey("b"), bytesValue("2"), DEFAULT_TIMESTAMP);
+
+ final ReadOnlyWindowStore<Bytes, byte[]> view =
cachingStore.readOnly(IsolationLevel.READ_UNCOMMITTED);
+ final List<KeyValue<Windowed<Bytes>, byte[]>> results =
toListAndCloseIterator(view.backwardAll());
+ assertEquals(2, results.size());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void
shouldReadUncommittedViewBackwardFetchAllMergesCacheAndStore(final boolean
hasIndex) {
+ setUp(hasIndex);
+ cachingStore.put(bytesKey("a"), bytesValue("1"), DEFAULT_TIMESTAMP);
+ cachingStore.commit(Map.of());
+ cachingStore.put(bytesKey("b"), bytesValue("2"), DEFAULT_TIMESTAMP);
+
+ final ReadOnlyWindowStore<Bytes, byte[]> view =
cachingStore.readOnly(IsolationLevel.READ_UNCOMMITTED);
+ final List<KeyValue<Windowed<Bytes>, byte[]>> results =
toListAndCloseIterator(
+ view.backwardFetchAll(ofEpochMilli(DEFAULT_TIMESTAMP),
ofEpochMilli(DEFAULT_TIMESTAMP)));
+ assertEquals(2, results.size());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void shouldThrowOnNullInstantInViewFetch(final boolean hasIndex) {
+ setUp(hasIndex);
+ final ReadOnlyWindowStore<Bytes, byte[]> view =
cachingStore.readOnly(IsolationLevel.READ_UNCOMMITTED);
+ // underlying.fetch validates via ApiUtils.validateMillisecondInstant
before toEpochMilli() is reached
+ assertThrows(IllegalArgumentException.class, () ->
view.fetch(bytesKey("a"), null, ofEpochMilli(0)));
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void shouldThrowNpeOnNullIsolationLevel(final boolean hasIndex) {
+ setUp(hasIndex);
+ assertThrows(NullPointerException.class, () ->
cachingStore.readOnly(null));
+ }
+
@SuppressWarnings("unchecked")
private void setUpCloseTests() {
underlyingStore = mock(RocksDBTimeOrderedWindowStore.class);
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/internals/TimeOrderedWindowStoreTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/TimeOrderedWindowStoreTest.java
index 51d65a330ae..5028380b893 100644
---
a/streams/src/test/java/org/apache/kafka/streams/state/internals/TimeOrderedWindowStoreTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/state/internals/TimeOrderedWindowStoreTest.java
@@ -1225,7 +1225,8 @@ public class TimeOrderedWindowStoreTest {
assertThat(
messages,
hasItem("Returning empty iterator for fetch with invalid key
range: from > to." +
- " This may be due to serdes that don't preserve ordering
when lexicographically comparing the serialized bytes." +
+ " This may be due to range arguments set in the wrong
order, " +
+ "or serdes that don't preserve ordering when
lexicographically comparing the serialized bytes." +
" Note that the built-in numerical serdes do not follow
this for negative numbers")
);
}
