@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 shouldFetchSessionForTimeRange() {
        // Only for TimeFirstSessionKeySchema schema
        if (!(getBaseSchema() instanceof TimeFirstSessionKeySchema)) {
            return;
        }
        final String keyA = "a";
        final String keyB = "b";
        final String keyC = "c";

        final Window[] sessionWindows = new Window[4];
        sessionWindows[0] = new SessionWindow(100L, 100L);
        sessionWindows[1] = new SessionWindow(50L, 200L);
        sessionWindows[2] = new SessionWindow(200L, 300L);
        bytesStore.put(serializeKey(new Windowed<>(keyA, sessionWindows[0])), serializeValue(10));
        bytesStore.put(serializeKey(new Windowed<>(keyB, sessionWindows[1])), serializeValue(100));
        bytesStore.put(serializeKey(new Windowed<>(keyC, sessionWindows[2])), serializeValue(200));


        // Fetch point
        try (final KeyValueIterator<Bytes, byte[]> values = ((RocksDBTimeOrderedSessionSegmentedBytesStore) bytesStore).fetchSessions(100L, 100L)) {

            final List<KeyValue<Windowed<String>, Long>> expected = Collections.singletonList(
                KeyValue.pair(new Windowed<>(keyA, sessionWindows[0]), 10L)
            );

            assertEquals(expected, toList(values));
        }

        // Fetch partial boundary
        try (final KeyValueIterator<Bytes, byte[]> values = ((RocksDBTimeOrderedSessionSegmentedBytesStore) bytesStore).fetchSessions(100L, 200L)) {

            final List<KeyValue<Windowed<String>, Long>> expected = asList(
                KeyValue.pair(new Windowed<>(keyA, sessionWindows[0]), 10L),
                KeyValue.pair(new Windowed<>(keyB, sessionWindows[1]), 100L)
            );

            assertEquals(expected, toList(values));
        }

        // Fetch partial
        try (final KeyValueIterator<Bytes, byte[]> values = ((RocksDBTimeOrderedSessionSegmentedBytesStore) bytesStore).fetchSessions(99L, 201L)) {
            final List<KeyValue<Windowed<String>, Long>> expected = asList(
                KeyValue.pair(new Windowed<>(keyA, sessionWindows[0]), 10L),
                KeyValue.pair(new Windowed<>(keyB, sessionWindows[1]), 100L)
            );

            assertEquals(expected, toList(values));
        }

        // Fetch partial
        try (final KeyValueIterator<Bytes, byte[]> values = ((RocksDBTimeOrderedSessionSegmentedBytesStore) bytesStore).fetchSessions(101L, 199L)) {
            assertTrue(toList(values).isEmpty());
        }

        // Fetch all boundary
        try (final KeyValueIterator<Bytes, byte[]> values = ((RocksDBTimeOrderedSessionSegmentedBytesStore) bytesStore).fetchSessions(100L, 300L)) {

            final List<KeyValue<Windowed<String>, Long>> expected = asList(
                KeyValue.pair(new Windowed<>(keyA, sessionWindows[0]), 10L),
                KeyValue.pair(new Windowed<>(keyB, sessionWindows[1]), 100L),
                KeyValue.pair(new Windowed<>(keyC, sessionWindows[2]), 200L)
            );

            assertEquals(expected, toList(values));
        }

        // Fetch all
        try (final KeyValueIterator<Bytes, byte[]> values = ((RocksDBTimeOrderedSessionSegmentedBytesStore) bytesStore).fetchSessions(99L, 301L)) {

            final List<KeyValue<Windowed<String>, Long>> expected = asList(
                KeyValue.pair(new Windowed<>(keyA, sessionWindows[0]), 10L),
                KeyValue.pair(new Windowed<>(keyB, sessionWindows[1]), 100L),
                KeyValue.pair(new Windowed<>(keyC, sessionWindows[2]), 200L)
            );

            assertEquals(expected, toList(values));
        }

        // Fetch all
        try (final KeyValueIterator<Bytes, byte[]> values = ((RocksDBTimeOrderedSessionSegmentedBytesStore) bytesStore).fetchSessions(101L, 299L)) {

            final List<KeyValue<Windowed<String>, Long>> expected = Collections.singletonList(
                KeyValue.pair(new Windowed<>(keyB, sessionWindows[1]), 100L)
            );

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