@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 shouldPutAndFetchEdgeKeyRange() {
        final String keyA = "a";
        final String keyB = "b";

        final Bytes serializedKeyAStart = serializeKey(new Windowed<>(keyA, startEdgeWindow), false,
            Integer.MAX_VALUE);
        final Bytes serializedKeyAEnd = serializeKey(new Windowed<>(keyA, endEdgeWindow), false,
            Integer.MAX_VALUE);
        final Bytes serializedKeyBStart = serializeKey(new Windowed<>(keyB, startEdgeWindow), false,
            Integer.MAX_VALUE);
        final Bytes serializedKeyBEnd = serializeKey(new Windowed<>(keyB, endEdgeWindow), false,
            Integer.MAX_VALUE);

        bytesStore.put(serializedKeyAStart, serializeValue(10));
        bytesStore.put(serializedKeyAEnd, serializeValue(50));
        bytesStore.put(serializedKeyBStart, serializeValue(100));
        bytesStore.put(serializedKeyBEnd, serializeValue(150));
        // Can fetch from start/end for key range
        try (final KeyValueIterator<Bytes, byte[]> values = bytesStore.fetch(
            Bytes.wrap(keyA.getBytes()), Bytes.wrap(keyB.getBytes()), startEdgeTime, endEdgeTime)) {

            final List<KeyValue<Windowed<String>, Long>> expected = getIndexSchema() == null ? asList(
                KeyValue.pair(new Windowed<>(keyA, startEdgeWindow), 10L),
                KeyValue.pair(new Windowed<>(keyB, startEdgeWindow), 100L),
                KeyValue.pair(new Windowed<>(keyA, endEdgeWindow), 50L),
                KeyValue.pair(new Windowed<>(keyB, endEdgeWindow), 150L)
            ) : asList(
                KeyValue.pair(new Windowed<>(keyA, startEdgeWindow), 10L),
                KeyValue.pair(new Windowed<>(keyA, endEdgeWindow), 50L),
                KeyValue.pair(new Windowed<>(keyB, startEdgeWindow), 100L),
                KeyValue.pair(new Windowed<>(keyB, endEdgeWindow), 150L)
            );
            assertEquals(expected, toList(values));
        }

        // Can fetch from 0 to max for key range
        try (final KeyValueIterator<Bytes, byte[]> values = bytesStore.fetch(
            Bytes.wrap(keyA.getBytes()), Bytes.wrap(keyB.getBytes()), 0L, Long.MAX_VALUE)) {

            final List<KeyValue<Windowed<String>, Long>> expected = getIndexSchema() == null ? asList(
                KeyValue.pair(new Windowed<>(keyA, startEdgeWindow), 10L),
                KeyValue.pair(new Windowed<>(keyB, startEdgeWindow), 100L),
                KeyValue.pair(new Windowed<>(keyA, endEdgeWindow), 50L),
                KeyValue.pair(new Windowed<>(keyB, endEdgeWindow), 150L)
            ) : asList(
                KeyValue.pair(new Windowed<>(keyA, startEdgeWindow), 10L),
                KeyValue.pair(new Windowed<>(keyA, endEdgeWindow), 50L),
                KeyValue.pair(new Windowed<>(keyB, startEdgeWindow), 100L),
                KeyValue.pair(new Windowed<>(keyB, endEdgeWindow), 150L)
            );
            assertEquals(expected, toList(values));
        }

        // KeyB should be ignored and KeyA should be included even in storage
        try (final KeyValueIterator<Bytes, byte[]> values = bytesStore.fetch(
            null, Bytes.wrap(keyA.getBytes()), startEdgeTime, endEdgeTime - 1L)) {

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

            assertEquals(expected, toList(values));
        }

        try (final KeyValueIterator<Bytes, byte[]> values = bytesStore.fetch(
            Bytes.wrap(keyB.getBytes()), null, startEdgeTime + 1, endEdgeTime)) {

            final List<KeyValue<Windowed<String>, Long>> expected = asList(
                KeyValue.pair(new Windowed<>(keyB, endEdgeWindow), 150L)
            );

            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 = getIndexSchema() == null ? asList(
                KeyValue.pair(new Windowed<>(keyA, startEdgeWindow), 10L),
                KeyValue.pair(new Windowed<>(keyB, startEdgeWindow), 100L),
                KeyValue.pair(new Windowed<>(keyA, endEdgeWindow), 50L),
                KeyValue.pair(new Windowed<>(keyB, endEdgeWindow), 150L)
            ) : asList(
                KeyValue.pair(new Windowed<>(keyA, startEdgeWindow), 10L),
                KeyValue.pair(new Windowed<>(keyA, endEdgeWindow), 50L),
                KeyValue.pair(new Windowed<>(keyB, startEdgeWindow), 100L),
                KeyValue.pair(new Windowed<>(keyB, endEdgeWindow), 150L)
            );
            assertEquals(expected, toList(values));
        }

        try (final KeyValueIterator<Bytes, byte[]> values = bytesStore.fetch(
            null, null, startEdgeTime, endEdgeTime)) {

            final List<KeyValue<Windowed<String>, Long>> expected = getIndexSchema() == null ? asList(
                KeyValue.pair(new Windowed<>(keyA, startEdgeWindow), 10L),
                KeyValue.pair(new Windowed<>(keyB, startEdgeWindow), 100L),
                KeyValue.pair(new Windowed<>(keyA, endEdgeWindow), 50L),
                KeyValue.pair(new Windowed<>(keyB, endEdgeWindow), 150L)
            ) : asList(
                KeyValue.pair(new Windowed<>(keyA, startEdgeWindow), 10L),
                KeyValue.pair(new Windowed<>(keyA, endEdgeWindow), 50L),
                KeyValue.pair(new Windowed<>(keyB, startEdgeWindow), 100L),
                KeyValue.pair(new Windowed<>(keyB, endEdgeWindow), 150L)
            );

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