@Override
    public void put(final Bytes rawBaseKey,
                    final byte[] value) {
        final long timestamp = baseKeySchema.segmentTimestamp(rawBaseKey);
        observedStreamTime = Math.max(observedStreamTime, timestamp);
        final long segmentId = segments.segmentId(timestamp);
        final S segment = segments.getOrCreateSegmentIfLive(segmentId, context, observedStreamTime);

        if (segment == null) {
            expiredRecordSensor.record(1.0d, context.currentSystemTimeMs());
            LOG.warn("Skipping record for expired segment.");
        } else {
            synchronized (position) {
                StoreQueryUtils.updatePosition(position, stateStoreContext);

                // Put to index first so that if put to base failed, when we iterate index, we will
                // find no base value. If put to base first but putting to index fails, when we iterate
                // index, we can't find the key but if we iterate over base store, we can find the key
                // which lead to inconsistency.
                if (hasIndex()) {
                    final KeyValue<Bytes, byte[]> indexKeyValue = getIndexKeyValue(rawBaseKey, value);
                    segment.put(indexKeyValue.key, indexKeyValue.value);
                }
                segment.put(rawBaseKey, value);
            }
        }
    }
@Test
    public void shouldPutAndFetchWithPrefixKey() {
        // Only for TimeFirstWindowKeySchema schema
        if (!(getBaseSchema() instanceof TimeFirstWindowKeySchema)) {
            return;
        }
        final String keyA = "a";
        final String keyB = "aa";
        final String keyC = "aaa";

        final Window maxWindow = new TimeWindow(Long.MAX_VALUE - 1, Long.MAX_VALUE);
        final Bytes serializedKeyA = serializeKey(new Windowed<>(keyA, maxWindow), false, Integer.MAX_VALUE);
        final Bytes serializedKeyB = serializeKey(new Windowed<>(keyB, maxWindow), false, Integer.MAX_VALUE);
        final Bytes serializedKeyC = serializeKey(new Windowed<>(keyC, maxWindow), false, Integer.MAX_VALUE);

        // Key are in decrease order but base storage binary key are in increase order
        assertTrue(serializedKeyA.compareTo(serializedKeyB) > 0);
        assertTrue(serializedKeyB.compareTo(serializedKeyC) > 0);

        bytesStore.put(serializedKeyA, serializeValue(10));
        bytesStore.put(serializedKeyB, serializeValue(50));
        bytesStore.put(serializedKeyC, serializeValue(100));

        try (final KeyValueIterator<Bytes, byte[]> values = bytesStore.fetch(
            Bytes.wrap(keyA.getBytes()), 0, Long.MAX_VALUE)) {

            final List<KeyValue<Windowed<String>, Long>> expected = asList(
                KeyValue.pair(new Windowed<>(keyA, maxWindow), 10L)
            );

            assertEquals(expected, toList(values));
        }

        try (final KeyValueIterator<Bytes, byte[]> values = bytesStore.fetch(
            Bytes.wrap(keyA.getBytes()), Bytes.wrap(keyB.getBytes()), 0, Long.MAX_VALUE)) {

            final List<KeyValue<Windowed<String>, Long>> expected = asList(
                KeyValue.pair(new Windowed<>(keyB, maxWindow), 50L),
                KeyValue.pair(new Windowed<>(keyA, maxWindow), 10L)
            );

            assertEquals(expected, toList(values));
        }

        // KeyC should be ignored and KeyA should be included even in storage, KeyC is before KeyB
        // and KeyA is after KeyB
        try (final KeyValueIterator<Bytes, byte[]> values = bytesStore.fetch(
            null, Bytes.wrap(keyB.getBytes()), 0, Long.MAX_VALUE)) {

            final List<KeyValue<Windowed<String>, Long>> expected = asList(
                KeyValue.pair(new Windowed<>(keyB, maxWindow), 50L),
                KeyValue.pair(new Windowed<>(keyA, maxWindow), 10L)
            );

            assertEquals(expected, toList(values));
        }

        // KeyC should be included even in storage KeyC is before KeyB
        try (final KeyValueIterator<Bytes, byte[]> values = bytesStore.fetch(
            Bytes.wrap(keyB.getBytes()), null, 0, Long.MAX_VALUE)) {

            final List<KeyValue<Windowed<String>, Long>> expected = asList(
                KeyValue.pair(new Windowed<>(keyC, maxWindow), 100L),
                KeyValue.pair(new Windowed<>(keyB, maxWindow), 50L)
            );

            assertEquals(expected, toList(values));
        }

        try (final KeyValueIterator<Bytes, byte[]> values = bytesStore.fetch(
            null, null, 0, Long.MAX_VALUE)) {

            final List<KeyValue<Windowed<String>, Long>> expected = asList(
                KeyValue.pair(new Windowed<>(keyC, maxWindow), 100L),
                KeyValue.pair(new Windowed<>(keyB, maxWindow), 50L),
                KeyValue.pair(new Windowed<>(keyA, maxWindow), 10L)
            );

            assertEquals(expected, toList(values));
        }
    }