test caching store across segments

apache · Oct 7, 2020 · a875cf1 · a875cf1
1 parent 64a95c0
commit a875cf1
Showing 1 changed file with 36 additions and 1 deletion.
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingSessionStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingSessionStoreTest.java
@@ -316,8 +316,8 @@ public void shouldFetchCorrectlyAcrossSegments() {
         final Windowed<Bytes> a3 = new Windowed<>(keyA, new SessionWindow(SEGMENT_INTERVAL * 2, SEGMENT_INTERVAL * 2));
         cachingStore.put(a1, "1".getBytes());
         cachingStore.put(a2, "2".getBytes());
-        cachingStore.put(a3, "3".getBytes());
         cachingStore.flush();
+        cachingStore.put(a3, "3".getBytes());
         final KeyValueIterator<Windowed<Bytes>, byte[]> results =
             cachingStore.findSessions(keyA, 0, SEGMENT_INTERVAL * 2);
         assertEquals(a1, results.next().key);
@@ -326,6 +326,23 @@ public void shouldFetchCorrectlyAcrossSegments() {
         assertFalse(results.hasNext());
     }
 
+    @Test
+    public void shouldBackwardFetchCorrectlyAcrossSegments() {
+        final Windowed<Bytes> a1 = new Windowed<>(keyA, new SessionWindow(SEGMENT_INTERVAL * 0, SEGMENT_INTERVAL * 0));
+        final Windowed<Bytes> a2 = new Windowed<>(keyA, new SessionWindow(SEGMENT_INTERVAL * 1, SEGMENT_INTERVAL * 1));
+        final Windowed<Bytes> a3 = new Windowed<>(keyA, new SessionWindow(SEGMENT_INTERVAL * 2, SEGMENT_INTERVAL * 2));
+        cachingStore.put(a1, "1".getBytes());
+        cachingStore.put(a2, "2".getBytes());
+        cachingStore.flush();
+        cachingStore.put(a3, "3".getBytes());
+        final KeyValueIterator<Windowed<Bytes>, byte[]> results =
+            cachingStore.backwardFindSessions(keyA, 0, SEGMENT_INTERVAL * 2);
+        assertEquals(a3, results.next().key);
+        assertEquals(a2, results.next().key);
+        assertEquals(a1, results.next().key);
+        assertFalse(results.hasNext());
+    }
+
     @Test
     public void shouldFetchRangeCorrectlyAcrossSegments() {
         final Windowed<Bytes> a1 = new Windowed<>(keyA, new SessionWindow(SEGMENT_INTERVAL * 0, SEGMENT_INTERVAL * 0));
@@ -518,6 +535,24 @@ public void shouldReturnSameResultsForSingleKeyFindSessionsAndEqualKeyRangeFindS
         assertFalse(keyRangeIterator.hasNext());
     }
 
+    @Test
+    public void shouldReturnSameResultsForSingleKeyFindSessionsBackwardsAndEqualKeyRangeFindSessions() {
+        cachingStore.put(new Windowed<>(keyA, new SessionWindow(0, 1)), "1".getBytes());
+        cachingStore.put(new Windowed<>(keyAA, new SessionWindow(2, 3)), "2".getBytes());
+        cachingStore.put(new Windowed<>(keyAA, new SessionWindow(4, 5)), "3".getBytes());
+        cachingStore.put(new Windowed<>(keyB, new SessionWindow(6, 7)), "4".getBytes());
+
+        final KeyValueIterator<Windowed<Bytes>, byte[]> singleKeyIterator =
+            cachingStore.backwardFindSessions(keyAA, 0L, 10L);
+        final KeyValueIterator<Windowed<Bytes>, byte[]> keyRangeIterator =
+            cachingStore.backwardFindSessions(keyAA, keyAA, 0L, 10L);
+
+        assertEquals(singleKeyIterator.next(), keyRangeIterator.next());
+        assertEquals(singleKeyIterator.next(), keyRangeIterator.next());
+        assertFalse(singleKeyIterator.hasNext());
+        assertFalse(keyRangeIterator.hasNext());
+    }
+
     @Test
     public void shouldClearNamespaceCacheOnClose() {
         final Windowed<Bytes> a1 = new Windowed<>(keyA, new SessionWindow(0, 0));