@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 shouldFindValuesWithinRange() {
        final String key = "a";
        bytesStore.put(serializeKey(new Windowed<>(key, windows[0])), serializeValue(10));
        bytesStore.put(serializeKey(new Windowed<>(key, windows[1])), serializeValue(50));
        bytesStore.put(serializeKey(new Windowed<>(key, windows[2])), serializeValue(100));
        // actual from: observedStreamTime - retention + 1
        // retention = 1000
        try (final KeyValueIterator<Bytes, byte[]> results = bytesStore.fetch(Bytes.wrap(key.getBytes()), 1, 999)) {

            final List<KeyValue<Windowed<String>, Long>> expected;

            // actual from: observedStreamTime - retention + 1
            if (getBaseSchema() instanceof TimeFirstWindowKeySchema) {
                // For windowkeyschema, actual from is 1
                // observed stream time = 1000. actual from = (1000 - 1000 + 1)
                // and search happens in the range 1-2000
                expected = asList(
                        KeyValue.pair(new Windowed<>(key, windows[0]), 10L),
                        KeyValue.pair(new Windowed<>(key, windows[1]), 50L)
                );
            } else {
                // For session key schema, actual from is 501
                // observed stream time = 1500. actual from = (1500 - 1000 + 1)
                // and search happens in the range 501-2000 deeming first record as expired.
                expected = Collections.singletonList(KeyValue.pair(new Windowed<>(key, windows[1]), 50L));
            }

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