private static Bytes wrapPrefix(final Bytes noPrefixKey, final boolean upperRange) {
            // Need to scan from prefix even key is null
            if (noPrefixKey == null) {
                final byte prefix = upperRange ? KEY_FIRST_PREFIX + 1 : KEY_FIRST_PREFIX;
                final byte[] ret = ByteBuffer.allocate(PREFIX_SIZE)
                    .put(prefix)
                    .array();
                return Bytes.wrap(ret);
            }
            final byte[] ret = ByteBuffer.allocate(PREFIX_SIZE + noPrefixKey.get().length)
                .put(KEY_FIRST_PREFIX)
                .put(noPrefixKey.get())
                .array();
            return Bytes.wrap(ret);
        }
@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