### Question:
SessionKeySchema implements SegmentedBytesStore.KeySchema { @Override public Bytes upperRange(Bytes key, long to) { final byte[] maxSuffix = ByteBuffer.allocate(SUFFIX_SIZE) .putLong(to) .putLong(to) .array(); return OrderedBytes.upperRange(key, maxSuffix); }  @Override void init(final String topic); @Override Bytes upperRangeFixedSize(final Bytes key, final long to); @Override Bytes lowerRangeFixedSize(final Bytes key, final long from); @Override Bytes upperRange(Bytes key, long to); @Override Bytes lowerRange(Bytes key, long from); @Override long segmentTimestamp(final Bytes key); @Override HasNextCondition hasNextCondition(final Bytes binaryKeyFrom, final Bytes binaryKeyTo, final long from, final long to); @Override List<Segment> segmentsToSearch(final Segments segments, final long from, final long to);  }### Answer:
@Test public void testUpperBoundWithLargeTimestamps() throws Exception { Bytes upper = sessionKeySchema.upperRange(Bytes.wrap(new byte[]{0xA, 0xB, 0xC}), Long.MAX_VALUE); assertThat( "shorter key with max timestamp should be in range", upper.compareTo( SessionKeySerde.bytesToBinary( new Windowed<>( Bytes.wrap(new byte[]{0xA}), new SessionWindow(Long.MAX_VALUE, Long.MAX_VALUE)) ) ) >= 0 ); assertThat( "shorter key with max timestamp should be in range", upper.compareTo( SessionKeySerde.bytesToBinary( new Windowed<>( Bytes.wrap(new byte[]{0xA, 0xB}), new SessionWindow(Long.MAX_VALUE, Long.MAX_VALUE)) ) ) >= 0 ); assertThat(upper, equalTo(SessionKeySerde.bytesToBinary( new Windowed<>(Bytes.wrap(new byte[]{0xA}), new SessionWindow(Long.MAX_VALUE, Long.MAX_VALUE)))) ); }
@Test public void testUpperBoundWithKeyBytesLargerThanFirstTimestampByte() throws Exception { Bytes upper = sessionKeySchema.upperRange(Bytes.wrap(new byte[]{0xA, (byte) 0x8F, (byte) 0x9F}), Long.MAX_VALUE); assertThat( "shorter key with max timestamp should be in range", upper.compareTo( SessionKeySerde.bytesToBinary( new Windowed<>( Bytes.wrap(new byte[]{0xA, (byte) 0x8F}), new SessionWindow(Long.MAX_VALUE, Long.MAX_VALUE)) ) ) >= 0 ); assertThat(upper, equalTo(SessionKeySerde.bytesToBinary( new Windowed<>(Bytes.wrap(new byte[]{0xA, (byte) 0x8F, (byte) 0x9F}), new SessionWindow(Long.MAX_VALUE, Long.MAX_VALUE)))) ); }
@Test public void testUpperBoundWithZeroTimestamp() throws Exception { Bytes upper = sessionKeySchema.upperRange(Bytes.wrap(new byte[]{0xA, 0xB, 0xC}), 0); assertThat(upper, equalTo(SessionKeySerde.bytesToBinary( new Windowed<>(Bytes.wrap(new byte[]{0xA, 0xB, 0xC}), new SessionWindow(0, 0)))) ); }