[GitHub] [kafka] mjsax commented on a diff in pull request #12204: [9/N][Emit final] Emit final for session window aggregations

[GitBox]

mjsax commented on code in PR #12204:
URL: https://github.com/apache/kafka/pull/12204#discussion_r907880169


##########
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedSessionSegmentedBytesStore.java:
##########
@@ -71,6 +72,36 @@ public byte[] fetchSession(final Bytes key,
         ));
     }
 
+    public KeyValueIterator<Bytes, byte[]> fetchSessions(final long 
earliestSessionEndTime,
+                                                         final long 
latestSessionEndTime) {
+        final List<KeyValueSegment> searchSpace = 
segments.segments(earliestSessionEndTime, latestSessionEndTime, true);
+
+        // here we use both as lower range since we the search boundaries are 
both on the session time end;
+        // hence effectively our search range is [0 earliestSE] - [0, latestSE]
+        final Bytes binaryFrom = baseKeySchema.lowerRangeFixedSize(null, 
earliestSessionEndTime);
+        final Bytes binaryTo = baseKeySchema.lowerRangeFixedSize(null, 
latestSessionEndTime);
+
+        return new SegmentIterator<>(
+                searchSpace.iterator(),
+                iterator -> {
+                    while (iterator.hasNext()) {
+                        final Bytes bytes = iterator.peekNextKey();
+
+                        final Windowed<Bytes> windowedKey = 
TimeFirstSessionKeySchema.from(bytes);
+                        final long endTime = windowedKey.window().end();
+
+                        if (endTime <= latestSessionEndTime && endTime >= 
earliestSessionEndTime)

Review Comment:
   nit: add `{}` to block



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org