improve format based on feedback

apache · Oct 7, 2020 · 64a95c0 · 64a95c0
1 parent ae7e3ed
commit 64a95c0
Show file tree

Hide file tree

Showing 3 changed files with 37 additions and 20 deletions.
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingSessionStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingSessionStore.java
@@ -420,30 +420,43 @@ private long currentSegmentLastTime() {
         }
 
         private void getNextSegmentIterator() {
-            ++currentSegmentId;
-            lastSegmentId = cacheFunction.segmentId(maxObservedTimestamp);
+            if (forward) {
+                ++currentSegmentId;
+                lastSegmentId = cacheFunction.segmentId(maxObservedTimestamp);
 
-            if (currentSegmentId > lastSegmentId) {
-                current = null;
-                return;
-            }
+                if (currentSegmentId > lastSegmentId) {
+                    current = null;
+                    return;
+                }
 
-            setCacheKeyRange(currentSegmentBeginTime(), currentSegmentLastTime());
+                setCacheKeyRange(currentSegmentBeginTime(), currentSegmentLastTime());
+
+                current.close();
 
-            current.close();
-            if (forward) {
                 current = context.cache().range(cacheName, cacheKeyFrom, cacheKeyTo);
             } else {
+                --currentSegmentId;
+
+                if (currentSegmentId < lastSegmentId) {
+                    current = null;
+                    return;
+                }
+
+                setCacheKeyRange(currentSegmentBeginTime(), currentSegmentLastTime());
+
+                current.close();
+
                 current = context.cache().reverseRange(cacheName, cacheKeyFrom, cacheKeyTo);
             }
+
         }
 
         private void setCacheKeyRange(final long lowerRangeEndTime, final long upperRangeEndTime) {
             if (cacheFunction.segmentId(lowerRangeEndTime) != cacheFunction.segmentId(upperRangeEndTime)) {
                 throw new IllegalStateException("Error iterating over segments: segment interval has changed");
             }
 
-            if (keyFrom == keyTo) {
+            if (keyFrom.equals(keyTo)) {
                 cacheKeyFrom = cacheFunction.cacheKey(segmentLowerRangeFixedSize(keyFrom, lowerRangeEndTime));
                 cacheKeyTo = cacheFunction.cacheKey(segmentUpperRangeFixedSize(keyTo, upperRangeEndTime));
             } else {

diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemorySessionStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemorySessionStore.java
@@ -250,7 +250,8 @@ public KeyValueIterator<Windowed<Bytes>, byte[]> fetch(final Bytes key) {
         return registerNewIterator(
             key,
             key,
-            Long.MAX_VALUE, endTimeMap.entrySet().iterator(),
+            Long.MAX_VALUE,
+            endTimeMap.entrySet().iterator(),
             true);
     }
 
@@ -484,13 +485,16 @@ private boolean setInnerIterators() {
                     recordIterator = entries.iterator();
                 } else {
                     final Set<Entry<Long, byte[]>> entries;
-                    if (forward) entries = nextKeyEntry.getValue()
-                        .headMap(latestSessionStartTime, true)
-                        .descendingMap()
-                        .entrySet();
-                    else entries = nextKeyEntry.getValue()
-                        .headMap(latestSessionStartTime, true)
-                        .entrySet();
+                    if (forward) {
+                        entries = nextKeyEntry.getValue()
+                            .headMap(latestSessionStartTime, true)
+                            .descendingMap()
+                            .entrySet();
+                    } else {
+                        entries = nextKeyEntry.getValue()
+                            .headMap(latestSessionStartTime, true)
+                            .entrySet();
+                    }
                     recordIterator = entries.iterator();
                 }
 

diff --git a/...src/test/java/org/apache/kafka/streams/state/internals/AbstractSessionBytesStoreTest.java b/...src/test/java/org/apache/kafka/streams/state/internals/AbstractSessionBytesStoreTest.java
@@ -405,7 +405,7 @@ public void shouldFetchExactKeys() {
     @Test
     public void shouldBackwardFetchExactKeys() {
         sessionStore = buildSessionStore(0x7a00000000000000L, Serdes.String(), Serdes.Long());
-        sessionStore.init(context, sessionStore);
+        sessionStore.init((StateStoreContext) context, sessionStore);
 
         sessionStore.put(new Windowed<>("a", new SessionWindow(0, 0)), 1L);
         sessionStore.put(new Windowed<>("aa", new SessionWindow(0, 10)), 2L);
@@ -473,7 +473,7 @@ public void shouldBackwardFetchAndIterateOverExactBinaryKeys() {
         final SessionStore<Bytes, String> sessionStore =
             buildSessionStore(RETENTION_PERIOD, Serdes.Bytes(), Serdes.String());
 
-        sessionStore.init(context, sessionStore);
+        sessionStore.init((StateStoreContext) context, sessionStore);
 
         final Bytes key1 = Bytes.wrap(new byte[] {0});
         final Bytes key2 = Bytes.wrap(new byte[] {0, 0});