[9/N][Emit final] Emit final for session window aggregations

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

@@ -181,11 +181,6 @@ public void process(final Record<KIn, VIn> record) {
                         store.remove(session.key);
 
                         maybeForwardUpdate(session.key, session.value, null);
-                        /*
-                        tupleForwarder.maybeForward(
-                            record.withKey(session.key)
-                                .withValue(new Change<>(null, session.value)));
-                         */
                     }
                 }
 
@@ -194,11 +189,6 @@ public void process(final Record<KIn, VIn> record) {
                 store.put(sessionKey, agg);
 
                 maybeForwardUpdate(sessionKey, null, agg);
-                /*
-                tupleForwarder.maybeForward(
-                    record.withKey(sessionKey)
-                        .withValue(new Change<>(agg, null)));
-                 */
             }
 
             maybeForwardFinalResult(record, windowCloseTime);
@@ -254,7 +244,7 @@ private boolean shouldEmitFinal(final long windowCloseTime) {
         }
 
         private long emitRangeLowerBound() {
-            return lastEmitWindowCloseTime == ConsumerRecord.NO_TIMESTAMP ? 0L : Math.max(0L, lastEmitWindowCloseTime);
+            return Math.max(0L, lastEmitWindowCloseTime);
         }
 
         private long emitRangeUpperBound(final long windowCloseTime) {
@@ -264,7 +254,9 @@ private long emitRangeUpperBound(final long windowCloseTime) {
         }
 
         private boolean shouldRangeFetch(final long emitRangeLowerBound, final long emitRangeUpperBound) {
-            return emitRangeUpperBound > emitRangeLowerBound;
+            // since a session window could be a single point (i.e. [t, t]),
+            // we need to range fetch and emit even if the upper and lower bound are the same
+            return emitRangeUpperBound >= emitRangeLowerBound;
         }
 
         private void fetchAndEmit(final Record<KIn, VIn> record,

streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractReadWriteDecorator.java

@@ -32,6 +32,7 @@
 import org.apache.kafka.streams.state.WindowStoreIterator;
 import org.apache.kafka.streams.state.internals.WrappedStateStore;
 
+import java.time.Instant;
 import java.util.List;
 
 abstract class AbstractReadWriteDecorator<T extends StateStore, K, V> extends WrappedStateStore<T, K, V> {
@@ -259,6 +260,12 @@ public KeyValueIterator<Windowed<K>, AGG> findSessions(final K keyFrom,
             return wrapped().findSessions(keyFrom, keyTo, earliestSessionEndTime, latestSessionStartTime);
         }
 
+        @Override
+        public KeyValueIterator<Windowed<K>, AGG> findSessions(final Instant earliestSessionEndTime,
+                                                               final Instant latestSessionEndTime) {
+            return wrapped().findSessions(earliestSessionEndTime, latestSessionEndTime);
+        }
+
         @Override
         public void remove(final Windowed<K> sessionKey) {
             wrapped().remove(sessionKey);

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

@@ -25,6 +25,8 @@
 import org.apache.kafka.streams.state.KeyValueIterator;
 import org.apache.kafka.streams.state.SessionStore;
 
+import java.time.Instant;
+
 import static org.apache.kafka.streams.processor.internals.ProcessorContextUtils.asInternalProcessorContext;
 
 /**
@@ -95,6 +97,12 @@ public byte[] fetchSession(final Bytes key, final long earliestSessionEndTime, f
         return wrapped().fetchSession(key, earliestSessionEndTime, latestSessionStartTime);
     }
 
+    @Override
+    public KeyValueIterator<Windowed<Bytes>, byte[]> findSessions(final Instant earliestSessionEndTime,
+                                                                  final Instant latestSessionEndTime) {
+        return wrapped().findSessions(earliestSessionEndTime, latestSessionEndTime);
+    }
+
     @Override
     public KeyValueIterator<Windowed<Bytes>, byte[]> backwardFetch(final Bytes key) {
         return wrapped().backwardFetch(key);

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

@@ -45,6 +45,7 @@
 import org.apache.kafka.streams.state.internals.StoreQueryUtils.QueryHandler;
 import org.apache.kafka.streams.state.internals.metrics.StateStoreMetrics;
 
+import java.time.Instant;
 import java.util.Map;
 import java.util.Objects;
 
@@ -346,6 +347,18 @@ public KeyValueIterator<Windowed<K>, V> findSessions(final K keyFrom,
             time);
     }
 
+    @Override
+    public KeyValueIterator<Windowed<K>, V> findSessions(final Instant earliestSessionEndTime,
+                                                         final Instant latestSessionEndTime) {
+        return new MeteredWindowedKeyValueIterator<>(
+                wrapped().findSessions(earliestSessionEndTime, latestSessionEndTime),
+                fetchSensor,
+                streamsMetrics,
+                serdes::keyFrom,
+                serdes::valueFrom,
+                time);
+    }
+
     @Override
     public KeyValueIterator<Windowed<K>, V> backwardFindSessions(final K keyFrom,
                                                                  final K keyTo,

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

@@ -203,8 +203,9 @@ public static void writeBinary(final ByteBuffer buf,
                                        final long endTime) {
             buf.putLong(endTime);
             buf.putLong(startTime);
-            if (key != null)
+            if (key != null) {
                 buf.put(key.get());
+            }
         }
 
         public static Bytes toBinary(final Bytes key,

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

@@ -39,7 +39,7 @@
  */
 public class RocksDBTimeOrderedSessionSegmentedBytesStore extends AbstractRocksDBTimeOrderedSegmentedBytesStore {
 
-    private class SessionKeySchemaIndexToBaseStoreIterator  extends IndexToBaseStoreIterator {
+    private class SessionKeySchemaIndexToBaseStoreIterator extends IndexToBaseStoreIterator {
         SessionKeySchemaIndexToBaseStoreIterator(final KeyValueIterator<Bytes, byte[]> indexIterator) {
             super(indexIterator);
         }
@@ -76,10 +76,10 @@ public KeyValueIterator<Bytes, byte[]> fetchSessions(final long earliestSessionE
                                                          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]
+        // here we want [0, latestSE, FF] as the upper bound to cover any possible keys,
+        // but since we can only get upper bound based on timestamps, we use a slight larger upper bound as [0, latestSE+1]
         final Bytes binaryFrom = baseKeySchema.lowerRangeFixedSize(null, earliestSessionEndTime);
-        final Bytes binaryTo = baseKeySchema.lowerRangeFixedSize(null, latestSessionEndTime);
+        final Bytes binaryTo = baseKeySchema.lowerRangeFixedSize(null, latestSessionEndTime + 1);
 
         return new SegmentIterator<>(
                 searchSpace.iterator(),
@@ -90,9 +90,9 @@ public KeyValueIterator<Bytes, byte[]> fetchSessions(final long earliestSessionE
                         final Windowed<Bytes> windowedKey = TimeFirstSessionKeySchema.from(bytes);
                         final long endTime = windowedKey.window().end();
 
-                        if (endTime <= latestSessionEndTime && endTime >= earliestSessionEndTime)
+                        if (endTime <= latestSessionEndTime && endTime >= earliestSessionEndTime) {
                             return true;
-
+                        }
                         iterator.next();
                     }
                     return false;

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

@@ -54,24 +54,18 @@ private SessionStore<Bytes, byte[]> maybeWrapCaching(final SessionStore<Bytes, b
             return inner;
         }
 
+        if (!inner.persistent()) {
+            return inner;
+        }
+
         // do not enable cache if the underlying store is time ordered
-        if (isTimeOrderedStore(inner)) {
+        if (storeSupplier instanceof RocksDbTimeOrderedSessionBytesStoreSupplier) {
             return inner;
         }
 
         return new CachingSessionStore(inner, storeSupplier.segmentIntervalMs());
     }
 
-    private boolean isTimeOrderedStore(final StateStore stateStore) {
-        if (stateStore instanceof RocksDBTimeOrderedWindowStore) {
-            return true;
-        }
-        if (stateStore instanceof WrappedStateStore) {
-            return isTimeOrderedStore(((WrappedStateStore) stateStore).wrapped());
-        }
-        return false;
-    }
-
     private SessionStore<Bytes, byte[]> maybeWrapLogging(final SessionStore<Bytes, byte[]> inner) {
         if (!enableLogging) {
             return inner;