public KeyValueIterator<Bytes, byte[]> fetchSessions(final long earliestSessionEndTime,
                                                         final long latestSessionEndTime) {
        final List<KeyValueSegment> searchSpace = segments.segments(earliestSessionEndTime, latestSessionEndTime, true);

        // 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 + 1);

        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) {
                            return true;
                        }
                        iterator.next();
                    }
                    return false;
                },
                binaryFrom,
                binaryTo,
                true);
    }
A RocksDB backed time-ordered segmented bytes store for session key schema.
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedSessionSegmentedBytesStore.java