@Override
        public HasNextCondition hasNextCondition(final Bytes binaryKeyFrom,
                                                 final Bytes binaryKeyTo,
                                                 final long from,
                                                 final long to,
                                                 final boolean forward) {
            return iterator -> {
                while (iterator.hasNext()) {
                    final Bytes bytes = iterator.peekNextKey();
                    final byte prefix = extractPrefix(bytes.get());

                    if (prefix != KEY_FIRST_PREFIX) {
                        return false;
                    }

                    final Windowed<Bytes> windowedKey = from(bytes);
                    final long endTime = windowedKey.window().end();
                    final long startTime = windowedKey.window().start();

                    if ((binaryKeyFrom == null || windowedKey.key().compareTo(binaryKeyFrom) >= 0)
                        && (binaryKeyTo == null || windowedKey.key().compareTo(binaryKeyTo) <= 0)
                        && endTime >= from
                        && startTime <= to) {
                        return true;
                    }
                    iterator.next();
                }
                return false;
            };
        }
@param key the key in the range
@param from the earliest end timestamp in the range
streams/src/main/java/org/apache/kafka/streams/state/internals/PrefixedSessionKeySchemas.java