emit final sw fix

streams/src/main/java/org/apache/kafka/streams/kstream/internals/SessionWindowedKStreamImpl.java

@@ -289,7 +289,10 @@ private <VR> StoreBuilder<SessionStore<K, VR>> materialize(final MaterializedInt
         // do not enable cache if the emit final strategy is used
         if (materialized.cachingEnabled() && emitStrategy.type() != EmitStrategy.StrategyType.ON_WINDOW_CLOSE) {
             builder.withCachingEnabled();
+        } else {
+            builder.withCachingDisabled();
         }
+
         return builder;
     }
 

streams/src/main/java/org/apache/kafka/streams/kstream/internals/SlidingWindowedKStreamImpl.java

@@ -233,7 +233,7 @@ private <VR> StoreBuilder<TimestampedWindowStore<K, VR>> materialize(final Mater
                             Duration.ofMillis(retentionPeriod),
                             Duration.ofMillis(windows.timeDifferenceMs()),
                             false,
-                            false
+                            true
                         ) :
                         Stores.persistentTimestampedWindowStore(
                             materialized.storeName(),
@@ -258,7 +258,9 @@ private <VR> StoreBuilder<TimestampedWindowStore<K, VR>> materialize(final Mater
         } else {
             builder.withLoggingDisabled();
         }
-        if (materialized.cachingEnabled()) {
+
+        // do not enable cache if the emit final strategy is used
+        if (materialized.cachingEnabled() && emitStrategy.type() != StrategyType.ON_WINDOW_CLOSE) {
             builder.withCachingEnabled();
         } else {
             builder.withCachingDisabled();

streams/src/main/java/org/apache/kafka/streams/state/internals/PrefixedSessionKeySchemas.java

@@ -102,8 +102,8 @@ public long segmentTimestamp(final Bytes key) {
         @Override
         public HasNextCondition hasNextCondition(final Bytes binaryKeyFrom,
                                                  final Bytes binaryKeyTo,
-                                                 final long from,
-                                                 final long to,
+                                                 final long earliestWindowEndTime,
+                                                 final long latestWindowStartTime,
                                                  final boolean forward) {
             return iterator -> {
                 while (iterator.hasNext()) {
@@ -120,13 +120,13 @@ public HasNextCondition hasNextCondition(final Bytes binaryKeyFrom,
 
                     // We can return false directly here since keys are sorted by end time and if
                     // we get time smaller than `from`, there won't be time within range.
-                    if (!forward && endTime < from) {
+                    if (!forward && endTime < earliestWindowEndTime) {
                         return false;
                     }
 
                     if ((binaryKeyFrom == null || windowedKey.key().compareTo(binaryKeyFrom) >= 0)
                         && (binaryKeyTo == null || windowedKey.key().compareTo(binaryKeyTo) <= 0)
-                        && endTime >= from && startTime <= to) {
+                        && endTime >= earliestWindowEndTime && startTime <= latestWindowStartTime) {
                         return true;
                     }
                     iterator.next();
@@ -137,10 +137,10 @@ public HasNextCondition hasNextCondition(final Bytes binaryKeyFrom,
 
         @Override
         public <S extends Segment> List<S> segmentsToSearch(final Segments<S> segments,
-                                                            final long from,
-                                                            final long to,
+                                                            final long earliestWindowEndTime,
+                                                            final long latestWindowStartTime,
                                                             final boolean forward) {
-            return segments.segments(from, Long.MAX_VALUE, forward);
+            return segments.segments(earliestWindowEndTime, Long.MAX_VALUE, forward);
         }
 
         static long extractStartTimestamp(final byte[] binaryKey) {

streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedSessionSegmentedBytesStore.java

@@ -63,12 +63,12 @@ protected Bytes getBaseKey(final Bytes indexKey) {
     }
 
     public byte[] fetchSession(final Bytes key,
-                               final long earliestSessionEndTime,
-                               final long latestSessionStartTime) {
+                               final long sessionStartTime,
+                               final long sessionEndTime) {
         return get(TimeFirstSessionKeySchema.toBinary(
             key,
-            earliestSessionEndTime,
-            latestSessionStartTime
+            sessionStartTime,
+            sessionEndTime
         ));
     }
 

streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedSessionStore.java

@@ -122,12 +122,12 @@ public KeyValueIterator<Windowed<Bytes>, byte[]> backwardFindSessions(final Byte
 
     @Override
     public byte[] fetchSession(final Bytes key,
-                               final long earliestSessionEndTime,
-                               final long latestSessionStartTime) {
+                               final long sessionStartTime,
+                               final long sessiontEndTime) {
         return wrapped().fetchSession(
             key,
-            earliestSessionEndTime,
-            latestSessionStartTime
+            sessionStartTime,
+            sessiontEndTime
         );
     }
 

streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractDualSchemaRocksDBSegmentedBytesStoreTest.java

@@ -91,6 +91,7 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
@@ -827,6 +828,143 @@ public void shouldPutAndBackwardFetchWithPrefix() {
         }
     }
 
+    @Test
+    public void shouldFetchSessionForSingleKey() {
+        // Only for TimeFirstSessionKeySchema schema
+        if (!(getBaseSchema() instanceof TimeFirstSessionKeySchema)) {
+            return;
+        }
+
+        final String keyA = "a";
+        final String keyB = "b";
+        final String keyC = "c";
+
+        final StateSerdes<String, Long> stateSerdes = StateSerdes.withBuiltinTypes("dummy", String.class, Long.class);
+        final Bytes key1 = Bytes.wrap(stateSerdes.keySerializer().serialize("dummy", keyA));
+        final Bytes key2 = Bytes.wrap(stateSerdes.keySerializer().serialize("dummy", keyB));
+        final Bytes key3 = Bytes.wrap(stateSerdes.keySerializer().serialize("dummy", keyC));
+
+        final byte[] expectedValue1 = serializeValue(10);
+        final byte[] expectedValue2 = serializeValue(50);
+        final byte[] expectedValue3 = serializeValue(100);
+        final byte[] expectedValue4 = serializeValue(200);
+
+        bytesStore.put(serializeKey(new Windowed<>(keyA, windows[0])), expectedValue1);
+        bytesStore.put(serializeKey(new Windowed<>(keyA, windows[1])), expectedValue2);
+        bytesStore.put(serializeKey(new Windowed<>(keyB, windows[2])), expectedValue3);
+        bytesStore.put(serializeKey(new Windowed<>(keyC, windows[3])), expectedValue4);
+
+        final byte[] value1 = ((RocksDBTimeOrderedSessionSegmentedBytesStore) bytesStore).fetchSession(
+            key1, windows[0].start(), windows[0].end());
+        assertEquals(Bytes.wrap(value1), Bytes.wrap(expectedValue1));
+
+        final byte[] value2 = ((RocksDBTimeOrderedSessionSegmentedBytesStore) bytesStore).fetchSession(
+            key1, windows[1].start(), windows[1].end());
+        assertEquals(Bytes.wrap(value2), Bytes.wrap(expectedValue2));
+
+        final byte[] value3 = ((RocksDBTimeOrderedSessionSegmentedBytesStore) bytesStore).fetchSession(
+            key2, windows[2].start(), windows[2].end());
+        assertEquals(Bytes.wrap(value3), Bytes.wrap(expectedValue3));
+
+        final byte[] value4 = ((RocksDBTimeOrderedSessionSegmentedBytesStore) bytesStore).fetchSession(
+            key3, windows[3].start(), windows[3].end());
+        assertEquals(Bytes.wrap(value4), Bytes.wrap(expectedValue4));
+
+        final byte[] noValue = ((RocksDBTimeOrderedSessionSegmentedBytesStore) bytesStore).fetchSession(
+            key3, 2000, 3000);
+        assertNull(noValue);
+    }
+
+    @Test
+    public void shouldFetchSessionForTimeRange() {
+        // Only for TimeFirstSessionKeySchema schema
+        if (!(getBaseSchema() instanceof TimeFirstSessionKeySchema)) {
+            return;
+        }
+        final String keyA = "a";
+        final String keyB = "b";
+        final String keyC = "c";
+
+        final Window[] sessionWindows = new Window[4];
+        sessionWindows[0] = new SessionWindow(100L, 100L);
+        sessionWindows[1] = new SessionWindow(50L, 200L);
+        sessionWindows[2] = new SessionWindow(200L, 300L);
+        bytesStore.put(serializeKey(new Windowed<>(keyA, sessionWindows[0])), serializeValue(10));
+        bytesStore.put(serializeKey(new Windowed<>(keyB, sessionWindows[1])), serializeValue(100));
+        bytesStore.put(serializeKey(new Windowed<>(keyC, sessionWindows[2])), serializeValue(200));
+
+
+        // Fetch point
+        try (final KeyValueIterator<Bytes, byte[]> values = ((RocksDBTimeOrderedSessionSegmentedBytesStore) bytesStore).fetchSessions(100L, 100L)) {
+
+            final List<KeyValue<Windowed<String>, Long>> expected = Collections.singletonList(
+                KeyValue.pair(new Windowed<>(keyA, sessionWindows[0]), 10L)
+            );
+
+            assertEquals(expected, toList(values));
+        }
+
+        // Fetch partial boundary
+        try (final KeyValueIterator<Bytes, byte[]> values = ((RocksDBTimeOrderedSessionSegmentedBytesStore) bytesStore).fetchSessions(100L, 200L)) {
+
+            final List<KeyValue<Windowed<String>, Long>> expected = asList(
+                KeyValue.pair(new Windowed<>(keyA, sessionWindows[0]), 10L),
+                KeyValue.pair(new Windowed<>(keyB, sessionWindows[1]), 100L)
+            );
+
+            assertEquals(expected, toList(values));
+        }
+
+        // Fetch partial
+        try (final KeyValueIterator<Bytes, byte[]> values = ((RocksDBTimeOrderedSessionSegmentedBytesStore) bytesStore).fetchSessions(99L, 201L)) {
+            final List<KeyValue<Windowed<String>, Long>> expected = asList(
+                KeyValue.pair(new Windowed<>(keyA, sessionWindows[0]), 10L),
+                KeyValue.pair(new Windowed<>(keyB, sessionWindows[1]), 100L)
+            );
+
+            assertEquals(expected, toList(values));
+        }
+
+        // Fetch partial
+        try (final KeyValueIterator<Bytes, byte[]> values = ((RocksDBTimeOrderedSessionSegmentedBytesStore) bytesStore).fetchSessions(101L, 199L)) {
+            assertTrue(toList(values).isEmpty());
+        }
+
+        // Fetch all boundary
+        try (final KeyValueIterator<Bytes, byte[]> values = ((RocksDBTimeOrderedSessionSegmentedBytesStore) bytesStore).fetchSessions(100L, 300L)) {
+
+            final List<KeyValue<Windowed<String>, Long>> expected = asList(
+                KeyValue.pair(new Windowed<>(keyA, sessionWindows[0]), 10L),
+                KeyValue.pair(new Windowed<>(keyB, sessionWindows[1]), 100L),
+                KeyValue.pair(new Windowed<>(keyC, sessionWindows[2]), 200L)
+            );
+
+            assertEquals(expected, toList(values));
+        }
+
+        // Fetch all
+        try (final KeyValueIterator<Bytes, byte[]> values = ((RocksDBTimeOrderedSessionSegmentedBytesStore) bytesStore).fetchSessions(99L, 301L)) {
+
+            final List<KeyValue<Windowed<String>, Long>> expected = asList(
+                KeyValue.pair(new Windowed<>(keyA, sessionWindows[0]), 10L),
+                KeyValue.pair(new Windowed<>(keyB, sessionWindows[1]), 100L),
+                KeyValue.pair(new Windowed<>(keyC, sessionWindows[2]), 200L)
+            );
+
+            assertEquals(expected, toList(values));
+        }
+
+        // Fetch all
+        try (final KeyValueIterator<Bytes, byte[]> values = ((RocksDBTimeOrderedSessionSegmentedBytesStore) bytesStore).fetchSessions(101L, 299L)) {
+
+            final List<KeyValue<Windowed<String>, Long>> expected = Collections.singletonList(
+                KeyValue.pair(new Windowed<>(keyB, sessionWindows[1]), 100L)
+            );
+
+            assertEquals(expected, toList(values));
+        }
+    }
+
     @Test
     public void shouldSkipAndRemoveDanglingIndex() {
         final String keyA = "a";