content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements. See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License. You may obtain a copy of the License at
   *
   *    http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.kafka.streams.state.internals;
  
  import org.apache.kafka.clients.consumer.ConsumerConfig;
  import org.apache.kafka.common.header.internals.RecordHeaders;
  import org.apache.kafka.common.metrics.Metrics;
  import org.apache.kafka.common.serialization.IntegerSerializer;
  import org.apache.kafka.common.serialization.Serdes;
  import org.apache.kafka.common.serialization.StringDeserializer;
  import org.apache.kafka.common.serialization.StringSerializer;
  import org.apache.kafka.common.utils.Bytes;
  import org.apache.kafka.common.utils.LogCaptureAppender;
  import org.apache.kafka.common.utils.LogContext;
  import org.apache.kafka.streams.KeyValue;
  import org.apache.kafka.streams.StreamsBuilder;
  import org.apache.kafka.streams.StreamsConfig;
  import org.apache.kafka.streams.TestInputTopic;
  import org.apache.kafka.streams.TopologyTestDriver;
  import org.apache.kafka.streams.errors.InvalidStateStoreException;
  import org.apache.kafka.streams.kstream.Consumed;
  import org.apache.kafka.streams.kstream.TimeWindowedDeserializer;
  import org.apache.kafka.streams.kstream.Windowed;
  import org.apache.kafka.streams.kstream.internals.TimeWindow;
  import org.apache.kafka.streams.processor.api.Processor;
  import org.apache.kafka.streams.processor.api.ProcessorContext;
  import org.apache.kafka.streams.processor.api.Record;
  import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
  import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
  import org.apache.kafka.streams.query.Position;
  import org.apache.kafka.streams.state.KeyValueIterator;
  import org.apache.kafka.streams.state.StoreBuilder;
  import org.apache.kafka.streams.state.Stores;
  import org.apache.kafka.streams.state.WindowStore;
  import org.apache.kafka.streams.state.WindowStoreIterator;
  import org.apache.kafka.test.InternalMockProcessorContext;
  import org.apache.kafka.test.TestUtils;
  
  import org.junit.jupiter.api.AfterEach;
  import org.junit.jupiter.api.BeforeEach;
  import org.junit.jupiter.api.Test;
  import org.junit.jupiter.api.extension.ExtendWith;
  import org.mockito.InOrder;
  import org.mockito.junit.jupiter.MockitoExtension;
  import org.mockito.junit.jupiter.MockitoSettings;
  import org.mockito.quality.Strictness;
  
  import java.nio.charset.StandardCharsets;
  import java.time.Duration;
  import java.time.Instant;
  import java.util.Arrays;
  import java.util.List;
  import java.util.Properties;
  import java.util.UUID;
  
  import static java.time.Duration.ofHours;
  import static java.time.Duration.ofMinutes;
  import static java.time.Instant.ofEpochMilli;
  import static java.util.Arrays.asList;
  import static org.apache.kafka.common.utils.Utils.mkEntry;
  import static org.apache.kafka.common.utils.Utils.mkMap;
  import static org.apache.kafka.streams.state.internals.ThreadCacheTest.memoryCacheEntrySize;
  import static org.apache.kafka.test.StreamsTestUtils.toListAndCloseIterator;
  import static org.apache.kafka.test.StreamsTestUtils.verifyAllWindowedKeyValues;
  import static org.apache.kafka.test.StreamsTestUtils.verifyKeyValueList;
  import static org.apache.kafka.test.StreamsTestUtils.verifyWindowedKeyValue;
  import static org.hamcrest.CoreMatchers.equalTo;
  import static org.hamcrest.CoreMatchers.hasItem;
  import static org.hamcrest.MatcherAssert.assertThat;
  import static org.junit.jupiter.api.Assertions.assertArrayEquals;
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertFalse;
  import static org.junit.jupiter.api.Assertions.assertNull;
  import static org.junit.jupiter.api.Assertions.assertThrows;
  import static org.junit.jupiter.api.Assertions.assertTrue;
  import static org.mockito.Mockito.doThrow;
  import static org.mockito.Mockito.inOrder;
  import static org.mockito.Mockito.mock;
  import static org.mockito.Mockito.verify;
  import static org.mockito.Mockito.when;
  
  @ExtendWith(MockitoExtension.class)
  @MockitoSettings(strictness = Strictness.STRICT_STUBS)
  public class CachingPersistentWindowStoreTest {
  
      private static final int MAX_CACHE_SIZE_BYTES = 150;
      private static final long DEFAULT_TIMESTAMP = 10L;
      private static final Long WINDOW_SIZE = 10L;
      private static final long SEGMENT_INTERVAL = 100L;
      private static final String TOPIC = "topic";
      private static final String CACHE_NAMESPACE = "0_0-store-name";
  
      private InternalMockProcessorContext<?, ?> context;
      private RocksDBSegmentedBytesStore bytesStore;
      private WindowStore<Bytes, byte[]> underlyingStore;
      private CachingWindowStore cachingStore;
      private CacheFlushListenerStub<Windowed<String>, String> cacheListener;
      private ThreadCache cache;
      private WindowKeySchema keySchema;
  
      @BeforeEach
      public void setUp() {
          keySchema = new WindowKeySchema();
          ///KAFKA-12960: Adding a retention of 100 ms to make all test cases work as is.
          bytesStore = new RocksDBSegmentedBytesStore("test", "metrics-scope", 100, SEGMENT_INTERVAL, keySchema);
          underlyingStore = new RocksDBWindowStore(bytesStore, false, WINDOW_SIZE);
          final TimeWindowedDeserializer<String> keyDeserializer = new TimeWindowedDeserializer<>(new StringDeserializer(), WINDOW_SIZE);
          keyDeserializer.setIsChangelogTopic(true);
          cacheListener = new CacheFlushListenerStub<>(keyDeserializer, new StringDeserializer());
          cachingStore = new CachingWindowStore(underlyingStore, WINDOW_SIZE, SEGMENT_INTERVAL);
          cachingStore.setFlushListener(cacheListener, false);
          cache = new ThreadCache(new LogContext("testCache "), MAX_CACHE_SIZE_BYTES, new MockStreamsMetrics(new Metrics()));
          context = new InternalMockProcessorContext<>(TestUtils.tempDirectory(), null, null, null, cache);
          context.setRecordContext(new ProcessorRecordContext(DEFAULT_TIMESTAMP, 0, 0, TOPIC, new RecordHeaders()));
          cachingStore.init(context, cachingStore);
      }
  
      @AfterEach
      public void closeStore() {
          cachingStore.close();
      }
  
      @SuppressWarnings("unchecked")
      @Test
      public void shouldDelegateInit() {
          final WindowStore<Bytes, byte[]> inner = mock(WindowStore.class);
          final CachingWindowStore outer = new CachingWindowStore(inner, WINDOW_SIZE, SEGMENT_INTERVAL);
          when(inner.name()).thenReturn("store");
          outer.init(context, outer);
          verify(inner).init(context, outer);
      }
  
      @Test
      public void shouldNotReturnDuplicatesInRanges() {
          final StreamsBuilder builder = new StreamsBuilder();
  
          final StoreBuilder<WindowStore<String, String>> storeBuilder = Stores.windowStoreBuilder(
                  Stores.persistentWindowStore("store-name", ofHours(1L), ofMinutes(1L), false),
                  Serdes.String(),
                  Serdes.String())
              .withCachingEnabled();
  
          builder.addStateStore(storeBuilder);
  
          builder.stream(TOPIC,
              Consumed.with(Serdes.String(), Serdes.String()))
              .process(() -> new Processor<String, String, String, String>() {
                  private WindowStore<String, String> store;
                  private int numRecordsProcessed;
                  private ProcessorContext<String, String> context;
  
                  @Override
                  public void init(final ProcessorContext<String, String> processorContext) {
                      this.context = processorContext;
                      this.store = processorContext.getStateStore("store-name");
                      int count = 0;
  
                      try (final KeyValueIterator<Windowed<String>, String> all = store.all()) {
                          while (all.hasNext()) {
                              count++;
                              all.next();
                          }
                      }
  
                      assertThat(count, equalTo(0));
                  }
  
                  @Override
                  public void process(final Record<String, String> record) {
                      int count = 0;
  
                      try (final KeyValueIterator<Windowed<String>, String> all = store.all()) {
                          while (all.hasNext()) {
                              count++;
                              all.next();
                          }
                      }
  
                      assertThat(count, equalTo(numRecordsProcessed));
  
                      store.put(record.value(), record.value(), record.timestamp());
  
                      numRecordsProcessed++;
  
                      context.forward(record);
                  }
              }, "store-name");
  
          final Properties streamsConfiguration = new Properties();
          streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
          streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);
          streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);
          streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath());
          streamsConfiguration.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10 * 1000L);
  
          final Instant initialWallClockTime = Instant.ofEpochMilli(0L);
          final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), streamsConfiguration, initialWallClockTime);
  
          final TestInputTopic<String, String> inputTopic = driver.createInputTopic(TOPIC,
              new StringSerializer(),
              new StringSerializer(),
              initialWallClockTime,
              Duration.ZERO);
  
          for (int i = 0; i < 5; i++) {
              inputTopic.pipeInput(UUID.randomUUID().toString(), UUID.randomUUID().toString());
          }
          driver.advanceWallClockTime(Duration.ofSeconds(10));
          inputTopic.advanceTime(Duration.ofSeconds(10));
          for (int i = 0; i < 5; i++) {
              inputTopic.pipeInput(UUID.randomUUID().toString(), UUID.randomUUID().toString());
          }
          driver.advanceWallClockTime(Duration.ofSeconds(10));
          inputTopic.advanceTime(Duration.ofSeconds(10));
          for (int i = 0; i < 5; i++) {
              inputTopic.pipeInput(UUID.randomUUID().toString(), UUID.randomUUID().toString());
          }
          driver.advanceWallClockTime(Duration.ofSeconds(10));
          inputTopic.advanceTime(Duration.ofSeconds(10));
          for (int i = 0; i < 5; i++) {
              inputTopic.pipeInput(UUID.randomUUID().toString(), UUID.randomUUID().toString());
          }
  
          driver.close();
      }
  
      @Test
      public void shouldPutFetchFromCache() {
          cachingStore.put(bytesKey("a"), bytesValue("a"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("b"), bytesValue("b"), DEFAULT_TIMESTAMP);
  
          assertThat(cachingStore.fetch(bytesKey("a"), 10), equalTo(bytesValue("a")));
          assertThat(cachingStore.fetch(bytesKey("b"), 10), equalTo(bytesValue("b")));
          assertThat(cachingStore.fetch(bytesKey("c"), 10), equalTo(null));
          assertThat(cachingStore.fetch(bytesKey("a"), 0), equalTo(null));
  
          try (final WindowStoreIterator<byte[]> a = cachingStore.fetch(bytesKey("a"), ofEpochMilli(10), ofEpochMilli(10));
               final WindowStoreIterator<byte[]> b = cachingStore.fetch(bytesKey("b"), ofEpochMilli(10), ofEpochMilli(10))) {
              verifyKeyValue(a.next(), DEFAULT_TIMESTAMP, "a");
              verifyKeyValue(b.next(), DEFAULT_TIMESTAMP, "b");
              assertFalse(a.hasNext());
              assertFalse(b.hasNext());
              assertEquals(2, cache.size());
          }
      }
  
      @Test
      public void shouldMatchPositionAfterPutWithFlushListener() {
          cachingStore.setFlushListener(record -> { }, false);
          shouldMatchPositionAfterPut();
      }
  
      @Test
      public void shouldMatchPositionAfterPutWithoutFlushListener() {
          cachingStore.setFlushListener(null, false);
          shouldMatchPositionAfterPut();
      }
  
      private void shouldMatchPositionAfterPut() {
          context.setRecordContext(new ProcessorRecordContext(0, 1, 0, "", new RecordHeaders()));
          cachingStore.put(bytesKey("key1"), bytesValue("value1"), DEFAULT_TIMESTAMP);
          context.setRecordContext(new ProcessorRecordContext(0, 2, 0, "", new RecordHeaders()));
          cachingStore.put(bytesKey("key2"), bytesValue("value2"), DEFAULT_TIMESTAMP);
  
          // Position should correspond to the last record's context, not the current context.
          context.setRecordContext(
              new ProcessorRecordContext(0, 3, 0, "", new RecordHeaders())
          );
  
          // the caching window store doesn't maintain a separate
          // position because it never serves queries from the cache
          assertEquals(Position.emptyPosition(), cachingStore.getPosition());
          assertEquals(Position.emptyPosition(), underlyingStore.getPosition());
  
          cachingStore.flush();
  
          assertEquals(
              Position.fromMap(mkMap(mkEntry("", mkMap(mkEntry(0, 2L))))),
              cachingStore.getPosition()
          );
          assertEquals(
              Position.fromMap(mkMap(mkEntry("", mkMap(mkEntry(0, 2L))))),
              underlyingStore.getPosition()
          );
      }
  
      private void verifyKeyValue(final KeyValue<Long, byte[]> next,
                                  final long expectedKey,
                                  final String expectedValue) {
          assertThat(next.key, equalTo(expectedKey));
          assertThat(next.value, equalTo(bytesValue(expectedValue)));
      }
  
      private static byte[] bytesValue(final String value) {
          return value.getBytes();
      }
  
      private static Bytes bytesKey(final String key) {
          return Bytes.wrap(key.getBytes());
      }
  
      @SuppressWarnings("resource")
      private String stringFrom(final byte[] from) {
          return new StringDeserializer().deserialize("", from);
      }
  
      @Test
      public void shouldPutFetchRangeFromCache() {
          cachingStore.put(bytesKey("a"), bytesValue("a"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("b"), bytesValue("b"), DEFAULT_TIMESTAMP);
  
          try (final KeyValueIterator<Windowed<Bytes>, byte[]> iterator =
                   cachingStore.fetch(bytesKey("a"), bytesKey("b"), ofEpochMilli(DEFAULT_TIMESTAMP), ofEpochMilli(DEFAULT_TIMESTAMP))) {
              final List<Windowed<Bytes>> expectedKeys = Arrays.asList(
                  new Windowed<>(bytesKey("a"), new TimeWindow(DEFAULT_TIMESTAMP, DEFAULT_TIMESTAMP + WINDOW_SIZE)),
                  new Windowed<>(bytesKey("b"), new TimeWindow(DEFAULT_TIMESTAMP, DEFAULT_TIMESTAMP + WINDOW_SIZE))
              );
  
              final List<String> expectedValues = Arrays.asList("a", "b");
  
              verifyAllWindowedKeyValues(iterator, expectedKeys, expectedValues);
              assertEquals(2, cache.size());
          }
      }
  
      @Test
      public void shouldPutFetchRangeFromCacheForNullKeyFrom() {
          cachingStore.put(bytesKey("a"), bytesValue("a"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("b"), bytesValue("b"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("c"), bytesValue("c"), DEFAULT_TIMESTAMP + 10L);
          cachingStore.put(bytesKey("d"), bytesValue("d"), DEFAULT_TIMESTAMP + 20L);
          cachingStore.put(bytesKey("e"), bytesValue("e"), DEFAULT_TIMESTAMP + 20L);
  
          try (final KeyValueIterator<Windowed<Bytes>, byte[]> iterator =
                   cachingStore.fetch(null, bytesKey("d"), ofEpochMilli(DEFAULT_TIMESTAMP), ofEpochMilli(DEFAULT_TIMESTAMP + 20L))) {
              final List<Windowed<Bytes>> expectedKeys = Arrays.asList(
                  new Windowed<>(bytesKey("a"), new TimeWindow(DEFAULT_TIMESTAMP, DEFAULT_TIMESTAMP + WINDOW_SIZE)),
                  new Windowed<>(bytesKey("b"), new TimeWindow(DEFAULT_TIMESTAMP, DEFAULT_TIMESTAMP + WINDOW_SIZE)),
                  new Windowed<>(bytesKey("c"), new TimeWindow(DEFAULT_TIMESTAMP + 10L, DEFAULT_TIMESTAMP + 10L + WINDOW_SIZE)),
                  new Windowed<>(bytesKey("d"), new TimeWindow(DEFAULT_TIMESTAMP + 20L, DEFAULT_TIMESTAMP + 20L + WINDOW_SIZE))
              );
  
              final List<String> expectedValues = Arrays.asList("a", "b", "c", "d");
  
              verifyAllWindowedKeyValues(iterator, expectedKeys, expectedValues);
          }
      }
  
      @Test
      public void shouldPutFetchRangeFromCacheForNullKeyTo() {
          cachingStore.put(bytesKey("a"), bytesValue("a"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("b"), bytesValue("b"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("c"), bytesValue("c"), DEFAULT_TIMESTAMP + 10L);
          cachingStore.put(bytesKey("d"), bytesValue("d"), DEFAULT_TIMESTAMP + 20L);
          cachingStore.put(bytesKey("e"), bytesValue("e"), DEFAULT_TIMESTAMP + 20L);
  
          try (final KeyValueIterator<Windowed<Bytes>, byte[]> iterator =
                   cachingStore.fetch(bytesKey("b"), null, ofEpochMilli(DEFAULT_TIMESTAMP), ofEpochMilli(DEFAULT_TIMESTAMP + 20L))) {
              final List<Windowed<Bytes>> expectedKeys = Arrays.asList(
                  new Windowed<>(bytesKey("b"), new TimeWindow(DEFAULT_TIMESTAMP, DEFAULT_TIMESTAMP + WINDOW_SIZE)),
                  new Windowed<>(bytesKey("c"), new TimeWindow(DEFAULT_TIMESTAMP + 10L, DEFAULT_TIMESTAMP + 10L + WINDOW_SIZE)),
                  new Windowed<>(bytesKey("d"), new TimeWindow(DEFAULT_TIMESTAMP + 20L, DEFAULT_TIMESTAMP + 20L + WINDOW_SIZE)),
                  new Windowed<>(bytesKey("e"), new TimeWindow(DEFAULT_TIMESTAMP + 20L, DEFAULT_TIMESTAMP + 20L + WINDOW_SIZE))
              );
  
              final List<String> expectedValues = Arrays.asList("b", "c", "d", "e");
  
              verifyAllWindowedKeyValues(iterator, expectedKeys, expectedValues);
          }
      }
  
      @Test
      public void shouldPutFetchRangeFromCacheForNullKeyFromKeyTo() {
          cachingStore.put(bytesKey("a"), bytesValue("a"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("b"), bytesValue("b"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("c"), bytesValue("c"), DEFAULT_TIMESTAMP + 10L);
          cachingStore.put(bytesKey("d"), bytesValue("d"), DEFAULT_TIMESTAMP + 20L);
          cachingStore.put(bytesKey("e"), bytesValue("e"), DEFAULT_TIMESTAMP + 20L);
  
          try (final KeyValueIterator<Windowed<Bytes>, byte[]> iterator =
                   cachingStore.fetch(null, null, ofEpochMilli(DEFAULT_TIMESTAMP), ofEpochMilli(DEFAULT_TIMESTAMP + 20L))) {
              final List<Windowed<Bytes>> expectedKeys = Arrays.asList(
                  new Windowed<>(bytesKey("a"), new TimeWindow(DEFAULT_TIMESTAMP, DEFAULT_TIMESTAMP + WINDOW_SIZE)),
                  new Windowed<>(bytesKey("b"), new TimeWindow(DEFAULT_TIMESTAMP, DEFAULT_TIMESTAMP + WINDOW_SIZE)),
                  new Windowed<>(bytesKey("c"), new TimeWindow(DEFAULT_TIMESTAMP + 10L, DEFAULT_TIMESTAMP + 10L + WINDOW_SIZE)),
                  new Windowed<>(bytesKey("d"), new TimeWindow(DEFAULT_TIMESTAMP + 20L, DEFAULT_TIMESTAMP + 20L + WINDOW_SIZE)),
                  new Windowed<>(bytesKey("e"), new TimeWindow(DEFAULT_TIMESTAMP + 20L, DEFAULT_TIMESTAMP + 20L + WINDOW_SIZE))
              );
  
              final List<String> expectedValues = Arrays.asList("a", "b", "c", "d", "e");
  
              verifyAllWindowedKeyValues(iterator, expectedKeys, expectedValues);
          }
      }
  
      @Test
      public void shouldPutBackwardFetchRangeFromCacheForNullKeyFrom() {
          cachingStore.put(bytesKey("a"), bytesValue("a"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("b"), bytesValue("b"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("c"), bytesValue("c"), DEFAULT_TIMESTAMP + 10L);
          cachingStore.put(bytesKey("d"), bytesValue("d"), DEFAULT_TIMESTAMP + 20L);
          cachingStore.put(bytesKey("e"), bytesValue("e"), DEFAULT_TIMESTAMP + 20L);
  
          try (final KeyValueIterator<Windowed<Bytes>, byte[]> iterator =
                   cachingStore.backwardFetch(null, bytesKey("c"), ofEpochMilli(DEFAULT_TIMESTAMP), ofEpochMilli(DEFAULT_TIMESTAMP + 20L))) {
              final List<Windowed<Bytes>> expectedKeys = Arrays.asList(
                  new Windowed<>(bytesKey("c"), new TimeWindow(DEFAULT_TIMESTAMP + 10L, DEFAULT_TIMESTAMP + 10L + WINDOW_SIZE)),
                  new Windowed<>(bytesKey("b"), new TimeWindow(DEFAULT_TIMESTAMP, DEFAULT_TIMESTAMP + WINDOW_SIZE)),
                  new Windowed<>(bytesKey("a"), new TimeWindow(DEFAULT_TIMESTAMP, DEFAULT_TIMESTAMP + WINDOW_SIZE))
              );
  
              final List<String> expectedValues = Arrays.asList("c", "b", "a");
  
              verifyAllWindowedKeyValues(iterator, expectedKeys, expectedValues);
          }
      }
  
      @Test
      public void shouldPutBackwardFetchRangeFromCacheForNullKeyTo() {
          cachingStore.put(bytesKey("a"), bytesValue("a"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("b"), bytesValue("b"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("c"), bytesValue("c"), DEFAULT_TIMESTAMP + 10L);
          cachingStore.put(bytesKey("d"), bytesValue("d"), DEFAULT_TIMESTAMP + 20L);
          cachingStore.put(bytesKey("e"), bytesValue("e"), DEFAULT_TIMESTAMP + 20L);
  
          try (final KeyValueIterator<Windowed<Bytes>, byte[]> iterator =
                   cachingStore.backwardFetch(bytesKey("c"), null, ofEpochMilli(DEFAULT_TIMESTAMP), ofEpochMilli(DEFAULT_TIMESTAMP + 20L))) {
              final List<Windowed<Bytes>> expectedKeys = Arrays.asList(
                  new Windowed<>(bytesKey("e"), new TimeWindow(DEFAULT_TIMESTAMP + 20L, DEFAULT_TIMESTAMP + 20L + WINDOW_SIZE)),
                  new Windowed<>(bytesKey("d"), new TimeWindow(DEFAULT_TIMESTAMP + 20L, DEFAULT_TIMESTAMP + 20L + WINDOW_SIZE)),
                  new Windowed<>(bytesKey("c"), new TimeWindow(DEFAULT_TIMESTAMP + 10L, DEFAULT_TIMESTAMP + 10L + WINDOW_SIZE))
              );
  
              final List<String> expectedValues = Arrays.asList("e", "d", "c");
  
              verifyAllWindowedKeyValues(iterator, expectedKeys, expectedValues);
          }
      }
  
      @Test
      public void shouldPutBackwardFetchRangeFromCacheForNullKeyFromKeyTo() {
          cachingStore.put(bytesKey("a"), bytesValue("a"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("b"), bytesValue("b"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("c"), bytesValue("c"), DEFAULT_TIMESTAMP + 10L);
          cachingStore.put(bytesKey("d"), bytesValue("d"), DEFAULT_TIMESTAMP + 20L);
          cachingStore.put(bytesKey("e"), bytesValue("e"), DEFAULT_TIMESTAMP + 20L);
  
          try (final KeyValueIterator<Windowed<Bytes>, byte[]> iterator =
                   cachingStore.backwardFetch(null, null, ofEpochMilli(DEFAULT_TIMESTAMP), ofEpochMilli(DEFAULT_TIMESTAMP + 20L))) {
              final List<Windowed<Bytes>> expectedKeys = Arrays.asList(
                  new Windowed<>(bytesKey("e"), new TimeWindow(DEFAULT_TIMESTAMP + 20L, DEFAULT_TIMESTAMP + 20L + WINDOW_SIZE)),
                  new Windowed<>(bytesKey("d"), new TimeWindow(DEFAULT_TIMESTAMP + 20L, DEFAULT_TIMESTAMP + 20L + WINDOW_SIZE)),
                  new Windowed<>(bytesKey("c"), new TimeWindow(DEFAULT_TIMESTAMP + 10L, DEFAULT_TIMESTAMP + 10L + WINDOW_SIZE)),
                  new Windowed<>(bytesKey("b"), new TimeWindow(DEFAULT_TIMESTAMP, DEFAULT_TIMESTAMP + WINDOW_SIZE)),
                  new Windowed<>(bytesKey("a"), new TimeWindow(DEFAULT_TIMESTAMP, DEFAULT_TIMESTAMP + WINDOW_SIZE))
              );
  
              final List<String> expectedValues = Arrays.asList("e", "d", "c", "b", "a");
  
              verifyAllWindowedKeyValues(iterator, expectedKeys, expectedValues);
          }
      }
  
      @Test
      public void shouldGetAllFromCache() {
          cachingStore.put(bytesKey("a"), bytesValue("a"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("b"), bytesValue("b"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("c"), bytesValue("c"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("d"), bytesValue("d"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("e"), bytesValue("e"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("f"), bytesValue("f"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("g"), bytesValue("g"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("h"), bytesValue("h"), DEFAULT_TIMESTAMP);
  
          try (final KeyValueIterator<Windowed<Bytes>, byte[]> iterator = cachingStore.all()) {
              final String[] array = {"a", "b", "c", "d", "e", "f", "g", "h"};
              for (final String s : array) {
                  verifyWindowedKeyValue(
                      iterator.next(),
                      new Windowed<>(bytesKey(s), new TimeWindow(DEFAULT_TIMESTAMP, DEFAULT_TIMESTAMP + WINDOW_SIZE)),
                      s);
              }
              assertFalse(iterator.hasNext());
          }
      }
  
      @Test
      public void shouldGetAllBackwardFromCache() {
          cachingStore.put(bytesKey("a"), bytesValue("a"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("b"), bytesValue("b"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("c"), bytesValue("c"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("d"), bytesValue("d"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("e"), bytesValue("e"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("f"), bytesValue("f"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("g"), bytesValue("g"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("h"), bytesValue("h"), DEFAULT_TIMESTAMP);
  
          try (final KeyValueIterator<Windowed<Bytes>, byte[]> iterator = cachingStore.backwardAll()) {
              final String[] array = {"h", "g", "f", "e", "d", "c", "b", "a"};
              for (final String s : array) {
                  verifyWindowedKeyValue(
                      iterator.next(),
                      new Windowed<>(bytesKey(s), new TimeWindow(DEFAULT_TIMESTAMP, DEFAULT_TIMESTAMP + WINDOW_SIZE)),
                      s);
              }
              assertFalse(iterator.hasNext());
          }
      }
  
      @Test
      public void shouldFetchAllWithinTimestampRange() {
          final String[] array = {"a", "b", "c", "d", "e", "f", "g", "h"};
          for (int i = 0; i < array.length; i++) {
              cachingStore.put(bytesKey(array[i]), bytesValue(array[i]), i);
          }
  
          try (final KeyValueIterator<Windowed<Bytes>, byte[]> iterator =
                   cachingStore.fetchAll(ofEpochMilli(0), ofEpochMilli(7))) {
              for (int i = 0; i < array.length; i++) {
                  final String str = array[i];
                  verifyWindowedKeyValue(
                      iterator.next(),
                      new Windowed<>(bytesKey(str), new TimeWindow(i, i + WINDOW_SIZE)),
                      str);
              }
              assertFalse(iterator.hasNext());
          }
  
          try (final KeyValueIterator<Windowed<Bytes>, byte[]> iterator1 =
                   cachingStore.fetchAll(ofEpochMilli(2), ofEpochMilli(4))) {
              for (int i = 2; i <= 4; i++) {
                  final String str = array[i];
                  verifyWindowedKeyValue(
                      iterator1.next(),
                      new Windowed<>(bytesKey(str), new TimeWindow(i, i + WINDOW_SIZE)),
                      str);
              }
              assertFalse(iterator1.hasNext());
          }
  
          try (final KeyValueIterator<Windowed<Bytes>, byte[]> iterator2 =
                   cachingStore.fetchAll(ofEpochMilli(5), ofEpochMilli(7))) {
              for (int i = 5; i <= 7; i++) {
                  final String str = array[i];
                  verifyWindowedKeyValue(
                      iterator2.next(),
                      new Windowed<>(bytesKey(str), new TimeWindow(i, i + WINDOW_SIZE)),
                      str);
              }
              assertFalse(iterator2.hasNext());
          }
      }
  
      @Test
      public void shouldFetchAllBackwardWithinTimestampRange() {
          final String[] array = {"a", "b", "c", "d", "e", "f", "g", "h"};
          for (int i = 0; i < array.length; i++) {
              cachingStore.put(bytesKey(array[i]), bytesValue(array[i]), i);
          }
  
          try (final KeyValueIterator<Windowed<Bytes>, byte[]> iterator =
                   cachingStore.backwardFetchAll(ofEpochMilli(0), ofEpochMilli(7))) {
              for (int i = array.length - 1; i >= 0; i--) {
                  final String str = array[i];
                  verifyWindowedKeyValue(
                      iterator.next(),
                      new Windowed<>(bytesKey(str), new TimeWindow(i, i + WINDOW_SIZE)),
                      str);
              }
              assertFalse(iterator.hasNext());
          }
  
          try (final KeyValueIterator<Windowed<Bytes>, byte[]> iterator1 =
                   cachingStore.backwardFetchAll(ofEpochMilli(2), ofEpochMilli(4))) {
              for (int i = 4; i >= 2; i--) {
                  final String str = array[i];
                  verifyWindowedKeyValue(
                      iterator1.next(),
                      new Windowed<>(bytesKey(str), new TimeWindow(i, i + WINDOW_SIZE)),
                      str);
              }
              assertFalse(iterator1.hasNext());
          }
  
          try (final KeyValueIterator<Windowed<Bytes>, byte[]> iterator2 =
                   cachingStore.backwardFetchAll(ofEpochMilli(5), ofEpochMilli(7))) {
              for (int i = 7; i >= 5; i--) {
                  final String str = array[i];
                  verifyWindowedKeyValue(
                      iterator2.next(),
                      new Windowed<>(bytesKey(str), new TimeWindow(i, i + WINDOW_SIZE)),
                      str);
              }
              assertFalse(iterator2.hasNext());
          }
      }
  
      @Test
      public void shouldFlushEvictedItemsIntoUnderlyingStore() {
          final int added = addItemsToCache();
          // all dirty entries should have been flushed
          try (final KeyValueIterator<Bytes, byte[]> iter = bytesStore.fetch(
              Bytes.wrap("0".getBytes(StandardCharsets.UTF_8)),
              DEFAULT_TIMESTAMP,
              DEFAULT_TIMESTAMP)) {
              final KeyValue<Bytes, byte[]> next = iter.next();
              assertEquals(DEFAULT_TIMESTAMP, keySchema.segmentTimestamp(next.key));
              assertArrayEquals("0".getBytes(), next.value);
              assertFalse(iter.hasNext());
              assertEquals(added - 1, cache.size());
          }
      }
  
      @Test
      public void shouldForwardDirtyItemsWhenFlushCalled() {
          final Windowed<String> windowedKey =
              new Windowed<>("1", new TimeWindow(DEFAULT_TIMESTAMP, DEFAULT_TIMESTAMP + WINDOW_SIZE));
          cachingStore.put(bytesKey("1"), bytesValue("a"), DEFAULT_TIMESTAMP);
          cachingStore.flush();
          assertEquals("a", cacheListener.forwarded.get(windowedKey).newValue);
          assertNull(cacheListener.forwarded.get(windowedKey).oldValue);
      }
  
      @Test
      public void shouldSetFlushListener() {
          assertTrue(cachingStore.setFlushListener(null, true));
          assertTrue(cachingStore.setFlushListener(null, false));
      }
  
      @Test
      public void shouldForwardOldValuesWhenEnabled() {
          cachingStore.setFlushListener(cacheListener, true);
          final Windowed<String> windowedKey =
              new Windowed<>("1", new TimeWindow(DEFAULT_TIMESTAMP, DEFAULT_TIMESTAMP + WINDOW_SIZE));
          cachingStore.put(bytesKey("1"), bytesValue("a"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("1"), bytesValue("b"), DEFAULT_TIMESTAMP);
          cachingStore.flush();
          assertEquals("b", cacheListener.forwarded.get(windowedKey).newValue);
          assertNull(cacheListener.forwarded.get(windowedKey).oldValue);
          cacheListener.forwarded.clear();
          cachingStore.put(bytesKey("1"), bytesValue("c"), DEFAULT_TIMESTAMP);
          cachingStore.flush();
          assertEquals("c", cacheListener.forwarded.get(windowedKey).newValue);
          assertEquals("b", cacheListener.forwarded.get(windowedKey).oldValue);
          cachingStore.put(bytesKey("1"), null, DEFAULT_TIMESTAMP);
          cachingStore.flush();
          assertNull(cacheListener.forwarded.get(windowedKey).newValue);
          assertEquals("c", cacheListener.forwarded.get(windowedKey).oldValue);
          cacheListener.forwarded.clear();
          cachingStore.put(bytesKey("1"), bytesValue("a"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("1"), bytesValue("b"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("1"), null, DEFAULT_TIMESTAMP);
          cachingStore.flush();
          assertNull(cacheListener.forwarded.get(windowedKey));
          cacheListener.forwarded.clear();
      }
  
      @Test
      public void shouldForwardOldValuesWhenDisabled() {
          final Windowed<String> windowedKey =
              new Windowed<>("1", new TimeWindow(DEFAULT_TIMESTAMP, DEFAULT_TIMESTAMP + WINDOW_SIZE));
          cachingStore.put(bytesKey("1"), bytesValue("a"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("1"), bytesValue("b"), DEFAULT_TIMESTAMP);
          cachingStore.flush();
          assertEquals("b", cacheListener.forwarded.get(windowedKey).newValue);
          assertNull(cacheListener.forwarded.get(windowedKey).oldValue);
          cachingStore.put(bytesKey("1"), bytesValue("c"), DEFAULT_TIMESTAMP);
          cachingStore.flush();
          assertEquals("c", cacheListener.forwarded.get(windowedKey).newValue);
          assertNull(cacheListener.forwarded.get(windowedKey).oldValue);
          cachingStore.put(bytesKey("1"), null, DEFAULT_TIMESTAMP);
          cachingStore.flush();
          assertNull(cacheListener.forwarded.get(windowedKey).newValue);
          assertNull(cacheListener.forwarded.get(windowedKey).oldValue);
          cacheListener.forwarded.clear();
          cachingStore.put(bytesKey("1"), bytesValue("a"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("1"), bytesValue("b"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("1"), null, DEFAULT_TIMESTAMP);
          cachingStore.flush();
          assertNull(cacheListener.forwarded.get(windowedKey));
          cacheListener.forwarded.clear();
      }
  
      @Test
      public void shouldForwardDirtyItemToListenerWhenEvicted() {
          final int numRecords = addItemsToCache();
          assertEquals(numRecords, cacheListener.forwarded.size());
      }
  
      @Test
      public void shouldTakeValueFromCacheIfSameTimestampFlushedToRocks() {
          cachingStore.put(bytesKey("1"), bytesValue("a"), DEFAULT_TIMESTAMP);
          cachingStore.flush();
          cachingStore.put(bytesKey("1"), bytesValue("b"), DEFAULT_TIMESTAMP);
  
          try (final WindowStoreIterator<byte[]> fetch =
                   cachingStore.fetch(bytesKey("1"), ofEpochMilli(DEFAULT_TIMESTAMP), ofEpochMilli(DEFAULT_TIMESTAMP))) {
              verifyKeyValue(fetch.next(), DEFAULT_TIMESTAMP, "b");
              assertFalse(fetch.hasNext());
          }
      }
  
      @Test
      public void shouldIterateAcrossWindows() {
          cachingStore.put(bytesKey("1"), bytesValue("a"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("1"), bytesValue("b"), DEFAULT_TIMESTAMP + WINDOW_SIZE);
  
          try (final WindowStoreIterator<byte[]> fetch =
                   cachingStore.fetch(bytesKey("1"), ofEpochMilli(DEFAULT_TIMESTAMP), ofEpochMilli(DEFAULT_TIMESTAMP + WINDOW_SIZE))) {
              verifyKeyValue(fetch.next(), DEFAULT_TIMESTAMP, "a");
              verifyKeyValue(fetch.next(), DEFAULT_TIMESTAMP + WINDOW_SIZE, "b");
              assertFalse(fetch.hasNext());
          }
      }
  
      @Test
      public void shouldIterateBackwardAcrossWindows() {
          cachingStore.put(bytesKey("1"), bytesValue("a"), DEFAULT_TIMESTAMP);
          cachingStore.put(bytesKey("1"), bytesValue("b"), DEFAULT_TIMESTAMP + WINDOW_SIZE);
  
          try (final WindowStoreIterator<byte[]> fetch =
                   cachingStore.backwardFetch(bytesKey("1"), ofEpochMilli(DEFAULT_TIMESTAMP), ofEpochMilli(DEFAULT_TIMESTAMP + WINDOW_SIZE))) {
              verifyKeyValue(fetch.next(), DEFAULT_TIMESTAMP + WINDOW_SIZE, "b");
              verifyKeyValue(fetch.next(), DEFAULT_TIMESTAMP, "a");
              assertFalse(fetch.hasNext());
          }
      }
  
      @Test
      public void shouldIterateCacheAndStore() {
          final Bytes key = Bytes.wrap("1".getBytes());
          bytesStore.put(WindowKeySchema.toStoreKeyBinary(key, DEFAULT_TIMESTAMP, 0), "a".getBytes());
          cachingStore.put(key, bytesValue("b"), DEFAULT_TIMESTAMP + WINDOW_SIZE);
          try (final WindowStoreIterator<byte[]> fetch =
                   cachingStore.fetch(bytesKey("1"), ofEpochMilli(DEFAULT_TIMESTAMP), ofEpochMilli(DEFAULT_TIMESTAMP + WINDOW_SIZE))) {
              verifyKeyValue(fetch.next(), DEFAULT_TIMESTAMP, "a");
              verifyKeyValue(fetch.next(), DEFAULT_TIMESTAMP + WINDOW_SIZE, "b");
              assertFalse(fetch.hasNext());
          }
      }
  
      @Test
      public void shouldIterateBackwardCacheAndStore() {
          final Bytes key = Bytes.wrap("1".getBytes());
          bytesStore.put(WindowKeySchema.toStoreKeyBinary(key, DEFAULT_TIMESTAMP, 0), "a".getBytes());
          cachingStore.put(key, bytesValue("b"), DEFAULT_TIMESTAMP + WINDOW_SIZE);
          try (final WindowStoreIterator<byte[]> fetch =
                   cachingStore.backwardFetch(bytesKey("1"), ofEpochMilli(DEFAULT_TIMESTAMP), ofEpochMilli(DEFAULT_TIMESTAMP + WINDOW_SIZE))) {
              verifyKeyValue(fetch.next(), DEFAULT_TIMESTAMP + WINDOW_SIZE, "b");
              verifyKeyValue(fetch.next(), DEFAULT_TIMESTAMP, "a");
              assertFalse(fetch.hasNext());
          }
      }
  
      @Test
      public void shouldIterateCacheAndStoreKeyRange() {
          final Bytes key = Bytes.wrap("1".getBytes());
          bytesStore.put(WindowKeySchema.toStoreKeyBinary(key, DEFAULT_TIMESTAMP, 0), "a".getBytes());
          cachingStore.put(key, bytesValue("b"), DEFAULT_TIMESTAMP + WINDOW_SIZE);
  
          try (final KeyValueIterator<Windowed<Bytes>, byte[]> fetchRange =
                   cachingStore.fetch(key, bytesKey("2"), ofEpochMilli(DEFAULT_TIMESTAMP), ofEpochMilli(DEFAULT_TIMESTAMP + WINDOW_SIZE))) {
              verifyWindowedKeyValue(
                  fetchRange.next(),
                  new Windowed<>(key, new TimeWindow(DEFAULT_TIMESTAMP, DEFAULT_TIMESTAMP + WINDOW_SIZE)),
                  "a");
              verifyWindowedKeyValue(
                  fetchRange.next(),
                  new Windowed<>(key, new TimeWindow(DEFAULT_TIMESTAMP + WINDOW_SIZE, DEFAULT_TIMESTAMP + WINDOW_SIZE + WINDOW_SIZE)),
                  "b");
              assertFalse(fetchRange.hasNext());
          }
      }
  
      @Test
      public void shouldIterateBackwardCacheAndStoreKeyRange() {
          final Bytes key = Bytes.wrap("1".getBytes());
          bytesStore.put(WindowKeySchema.toStoreKeyBinary(key, DEFAULT_TIMESTAMP, 0), "a".getBytes());
          cachingStore.put(key, bytesValue("b"), DEFAULT_TIMESTAMP + WINDOW_SIZE);
  
          try (final KeyValueIterator<Windowed<Bytes>, byte[]> fetchRange =
                   cachingStore.backwardFetch(key, bytesKey("2"), ofEpochMilli(DEFAULT_TIMESTAMP), ofEpochMilli(DEFAULT_TIMESTAMP + WINDOW_SIZE))) {
              verifyWindowedKeyValue(
                  fetchRange.next(),
                  new Windowed<>(key, new TimeWindow(DEFAULT_TIMESTAMP + WINDOW_SIZE, DEFAULT_TIMESTAMP + WINDOW_SIZE + WINDOW_SIZE)),
                  "b");
              verifyWindowedKeyValue(
                  fetchRange.next(),
                  new Windowed<>(key, new TimeWindow(DEFAULT_TIMESTAMP, DEFAULT_TIMESTAMP + WINDOW_SIZE)),
                  "a");
              assertFalse(fetchRange.hasNext());
          }
      }
  
      @Test
      public void shouldClearNamespaceCacheOnClose() {
          cachingStore.put(bytesKey("a"), bytesValue("a"), 0L);
          assertEquals(1, cache.size());
          cachingStore.close();
          assertEquals(0, cache.size());
      }
  
      @SuppressWarnings("resource")
      @Test
      public void shouldThrowIfTryingToFetchFromClosedCachingStore() {
          cachingStore.close();
          assertThrows(InvalidStateStoreException.class, () -> cachingStore.fetch(bytesKey("a"), ofEpochMilli(0), ofEpochMilli(10)));
      }
  
      @SuppressWarnings("resource")
      @Test
      public void shouldThrowIfTryingToFetchRangeFromClosedCachingStore() {
          cachingStore.close();
          assertThrows(InvalidStateStoreException.class, () -> cachingStore.fetch(bytesKey("a"), bytesKey("b"), ofEpochMilli(0), ofEpochMilli(10)));
      }
  
      @Test
      public void shouldThrowIfTryingToWriteToClosedCachingStore() {
          cachingStore.close();
          assertThrows(InvalidStateStoreException.class, () -> cachingStore.put(bytesKey("a"), bytesValue("a"), 0L));
      }
  
      @Test
      public void shouldFetchAndIterateOverExactKeys() {
          cachingStore.put(bytesKey("a"), bytesValue("0001"), 0);
          cachingStore.put(bytesKey("aa"), bytesValue("0002"), 0);
          cachingStore.put(bytesKey("a"), bytesValue("0003"), 1);
          cachingStore.put(bytesKey("aa"), bytesValue("0004"), 1);
          cachingStore.put(bytesKey("a"), bytesValue("0005"), SEGMENT_INTERVAL);
  
          final List<KeyValue<Long, byte[]>> expected = asList(
              KeyValue.pair(0L, bytesValue("0001")),
              KeyValue.pair(1L, bytesValue("0003")),
              KeyValue.pair(SEGMENT_INTERVAL, bytesValue("0005"))
          );
          final List<KeyValue<Long, byte[]>> actual =
              toListAndCloseIterator(cachingStore.fetch(bytesKey("a"), ofEpochMilli(0), ofEpochMilli(Long.MAX_VALUE)));
          verifyKeyValueList(expected, actual);
      }
  
      @Test
      public void shouldBackwardFetchAndIterateOverExactKeys() {
          cachingStore.put(bytesKey("a"), bytesValue("0001"), 0);
          cachingStore.put(bytesKey("aa"), bytesValue("0002"), 0);
          cachingStore.put(bytesKey("a"), bytesValue("0003"), 1);
          cachingStore.put(bytesKey("aa"), bytesValue("0004"), 1);
          cachingStore.put(bytesKey("a"), bytesValue("0005"), SEGMENT_INTERVAL);
  
          final List<KeyValue<Long, byte[]>> expected = asList(
              KeyValue.pair(SEGMENT_INTERVAL, bytesValue("0005")),
              KeyValue.pair(1L, bytesValue("0003")),
              KeyValue.pair(0L, bytesValue("0001"))
          );
          final List<KeyValue<Long, byte[]>> actual =
              toListAndCloseIterator(cachingStore.backwardFetch(bytesKey("a"), ofEpochMilli(0), ofEpochMilli(Long.MAX_VALUE)));
          verifyKeyValueList(expected, actual);
      }
  
      @Test
      public void shouldFetchAndIterateOverKeyRange() {
          cachingStore.put(bytesKey("a"), bytesValue("0001"), 0);
          cachingStore.put(bytesKey("aa"), bytesValue("0002"), 0);
          cachingStore.put(bytesKey("a"), bytesValue("0003"), 1);
          cachingStore.put(bytesKey("aa"), bytesValue("0004"), 1);
          cachingStore.put(bytesKey("a"), bytesValue("0005"), SEGMENT_INTERVAL);
  
          verifyKeyValueList(
              asList(
                  windowedPair("a", "0001", 0),
                  windowedPair("a", "0003", 1),
                  windowedPair("a", "0005", SEGMENT_INTERVAL)
              ),
              toListAndCloseIterator(cachingStore.fetch(bytesKey("a"), bytesKey("a"), ofEpochMilli(0), ofEpochMilli(Long.MAX_VALUE)))
          );
  
          verifyKeyValueList(
              asList(
                  windowedPair("aa", "0002", 0),
                  windowedPair("aa", "0004", 1)),
              toListAndCloseIterator(cachingStore.fetch(bytesKey("aa"), bytesKey("aa"), ofEpochMilli(0), ofEpochMilli(Long.MAX_VALUE)))
          );
  
          verifyKeyValueList(
              asList(
                  windowedPair("a", "0001", 0),
                  windowedPair("a", "0003", 1),
                  windowedPair("aa", "0002", 0),
                  windowedPair("aa", "0004", 1),
                  windowedPair("a", "0005", SEGMENT_INTERVAL)
              ),
              toListAndCloseIterator(cachingStore.fetch(bytesKey("a"), bytesKey("aa"), ofEpochMilli(0), ofEpochMilli(Long.MAX_VALUE)))
          );
      }
  
      @Test
      public void shouldFetchAndIterateOverKeyBackwardRange() {
          cachingStore.put(bytesKey("a"), bytesValue("0001"), 0);
          cachingStore.put(bytesKey("aa"), bytesValue("0002"), 0);
          cachingStore.put(bytesKey("a"), bytesValue("0003"), 1);
          cachingStore.put(bytesKey("aa"), bytesValue("0004"), 1);
          cachingStore.put(bytesKey("a"), bytesValue("0005"), SEGMENT_INTERVAL);
  
          verifyKeyValueList(
              asList(
                  windowedPair("a", "0005", SEGMENT_INTERVAL),
                  windowedPair("a", "0003", 1),
                  windowedPair("a", "0001", 0)
              ),
              toListAndCloseIterator(cachingStore.backwardFetch(bytesKey("a"), bytesKey("a"), ofEpochMilli(0), ofEpochMilli(Long.MAX_VALUE)))
          );
  
          verifyKeyValueList(
              asList(
                  windowedPair("aa", "0004", 1),
                  windowedPair("aa", "0002", 0)),
              toListAndCloseIterator(cachingStore.backwardFetch(bytesKey("aa"), bytesKey("aa"), ofEpochMilli(0), ofEpochMilli(Long.MAX_VALUE)))
          );
  
          verifyKeyValueList(
              asList(
                  windowedPair("a", "0005", SEGMENT_INTERVAL),
                  windowedPair("aa", "0004", 1),
                  windowedPair("aa", "0002", 0),
                  windowedPair("a", "0003", 1),
                  windowedPair("a", "0001", 0)
              ),
              toListAndCloseIterator(cachingStore.backwardFetch(bytesKey("a"), bytesKey("aa"), ofEpochMilli(0), ofEpochMilli(Long.MAX_VALUE)))
          );
      }
  
      @Test
      public void shouldReturnSameResultsForSingleKeyFetchAndEqualKeyRangeFetch() {
          cachingStore.put(bytesKey("a"), bytesValue("0001"), 0);
          cachingStore.put(bytesKey("aa"), bytesValue("0002"), 1);
          cachingStore.put(bytesKey("aa"), bytesValue("0003"), 2);
          cachingStore.put(bytesKey("aaa"), bytesValue("0004"), 3);
  
          try (final WindowStoreIterator<byte[]> singleKeyIterator = cachingStore.fetch(bytesKey("aa"), 0L, 5L);
               final KeyValueIterator<Windowed<Bytes>, byte[]> keyRangeIterator = cachingStore.fetch(bytesKey("aa"), bytesKey("aa"), 0L, 5L)) {
  
              assertEquals(stringFrom(singleKeyIterator.next().value), stringFrom(keyRangeIterator.next().value));
              assertEquals(stringFrom(singleKeyIterator.next().value), stringFrom(keyRangeIterator.next().value));
              assertFalse(singleKeyIterator.hasNext());
              assertFalse(keyRangeIterator.hasNext());
          }
      }
  
      @Test
      public void shouldReturnSameResultsForSingleKeyFetchAndEqualKeyRangeBackwardFetch() {
          cachingStore.put(bytesKey("a"), bytesValue("0001"), 0);
          cachingStore.put(bytesKey("aa"), bytesValue("0002"), 1);
          cachingStore.put(bytesKey("aa"), bytesValue("0003"), 2);
          cachingStore.put(bytesKey("aaa"), bytesValue("0004"), 3);
  
          try (final WindowStoreIterator<byte[]> singleKeyIterator =
                   cachingStore.backwardFetch(bytesKey("aa"), Instant.ofEpochMilli(0L), Instant.ofEpochMilli(5L));
               final KeyValueIterator<Windowed<Bytes>, byte[]> keyRangeIterator =
                   cachingStore.backwardFetch(bytesKey("aa"), bytesKey("aa"), Instant.ofEpochMilli(0L), Instant.ofEpochMilli(5L))) {
  
              assertEquals(stringFrom(singleKeyIterator.next().value), stringFrom(keyRangeIterator.next().value));
              assertEquals(stringFrom(singleKeyIterator.next().value), stringFrom(keyRangeIterator.next().value));
              assertFalse(singleKeyIterator.hasNext());
              assertFalse(keyRangeIterator.hasNext());
          }
      }
  
      @Test
      public void shouldThrowNullPointerExceptionOnPutNullKey() {
          assertThrows(NullPointerException.class, () -> cachingStore.put(null, bytesValue("anyValue"), 0L));
      }
  
      @Test
      public void shouldNotThrowNullPointerExceptionOnPutNullValue() {
          cachingStore.put(bytesKey("a"), null, 0L);
      }
  
      @SuppressWarnings("resource")
      @Test
      public void shouldThrowNullPointerExceptionOnFetchNullKey() {
          assertThrows(NullPointerException.class, () -> cachingStore.fetch(null, ofEpochMilli(1L), ofEpochMilli(2L)));
      }
  
      @SuppressWarnings("resource")
      @Test
      public void shouldNotThrowInvalidRangeExceptionWithNegativeFromKey() {
          final Bytes keyFrom = Bytes.wrap(new IntegerSerializer().serialize("", -1));
          final Bytes keyTo = Bytes.wrap(new IntegerSerializer().serialize("", 1));
  
          try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(CachingWindowStore.class);
               final KeyValueIterator<Windowed<Bytes>, byte[]> iterator = cachingStore.fetch(keyFrom, keyTo, 0L, 10L)) {
              assertFalse(iterator.hasNext());
  
              final List<String> messages = appender.getMessages();
              assertThat(
                  messages,
                  hasItem("Returning empty iterator for fetch with invalid key range: from > to." +
                      " This may be due to range arguments set in the wrong order, " +
                      "or serdes that don't preserve ordering when lexicographically comparing the serialized bytes." +
                      " Note that the built-in numerical serdes do not follow this for negative numbers")
              );
          }
      }
  
      @SuppressWarnings("resource")
      @Test
      public void shouldNotThrowInvalidBackwardRangeExceptionWithNegativeFromKey() {
          final Bytes keyFrom = Bytes.wrap(new IntegerSerializer().serialize("", -1));
          final Bytes keyTo = Bytes.wrap(new IntegerSerializer().serialize("", 1));
  
          try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(CachingWindowStore.class);
               final KeyValueIterator<Windowed<Bytes>, byte[]> iterator =
                   cachingStore.backwardFetch(keyFrom, keyTo, Instant.ofEpochMilli(0L), Instant.ofEpochMilli(10L))) {
              assertFalse(iterator.hasNext());
  
              final List<String> messages = appender.getMessages();
              assertThat(
                  messages,
                  hasItem("Returning empty iterator for fetch with invalid key range: from > to." +
                      " This may be due to serdes that don't preserve ordering when lexicographically comparing the serialized bytes." +
                      " Note that the built-in numerical serdes do not follow this for negative numbers")
              );
          }
      }
  
      @Test
      public void shouldCloseCacheAndWrappedStoreAfterErrorDuringCacheFlush() {
          setUpCloseTests();
          final InOrder inOrder = inOrder(cache, underlyingStore);
          doThrow(new RuntimeException("Simulating an error on flush")).doNothing().when(cache).flush(CACHE_NAMESPACE);
  
          assertThrows(RuntimeException.class, cachingStore::close);
          inOrder.verify(cache).close(CACHE_NAMESPACE);
          inOrder.verify(underlyingStore).close();
      }
  
      @Test
      public void shouldCloseWrappedStoreAfterErrorDuringCacheClose() {
          setUpCloseTests();
          final InOrder inOrder = inOrder(cache, underlyingStore);
          doThrow(new RuntimeException("Simulating an error on close")).doNothing().when(cache).close(CACHE_NAMESPACE);
  
          assertThrows(RuntimeException.class, cachingStore::close);
          inOrder.verify(cache).flush(CACHE_NAMESPACE);
          inOrder.verify(underlyingStore).close();
      }
  
      @Test
      public void shouldCloseCacheAfterErrorDuringStateStoreClose() {
          setUpCloseTests();
          final InOrder inOrder = inOrder(cache);
          doThrow(new RuntimeException("Simulating an error on close")).doNothing().when(underlyingStore).close();
  
          assertThrows(RuntimeException.class, cachingStore::close);
          inOrder.verify(cache).flush(CACHE_NAMESPACE);
          inOrder.verify(cache).close(CACHE_NAMESPACE);
      }
  
      @SuppressWarnings("unchecked")
      private void setUpCloseTests() {
          underlyingStore = mock(WindowStore.class);
          when(underlyingStore.name()).thenReturn("store-name");
          cachingStore = new CachingWindowStore(underlyingStore, WINDOW_SIZE, SEGMENT_INTERVAL);
          cache = mock(ThreadCache.class);
          context = new InternalMockProcessorContext<>(TestUtils.tempDirectory(), null, null, null, cache);
          context.setRecordContext(new ProcessorRecordContext(10, 0, 0, TOPIC, new RecordHeaders()));
          cachingStore.init(context, cachingStore);
      }
  
      private static KeyValue<Windowed<Bytes>, byte[]> windowedPair(final String key, final String value, final long timestamp) {
          return KeyValue.pair(
              new Windowed<>(bytesKey(key), new TimeWindow(timestamp, timestamp + WINDOW_SIZE)),
              bytesValue(value));
      }
  
      private int addItemsToCache() {
          long cachedSize = 0;
          int i = 0;
          while (cachedSize < MAX_CACHE_SIZE_BYTES) {
              final String kv = String.valueOf(i++);
              cachingStore.put(bytesKey(kv), bytesValue(kv), DEFAULT_TIMESTAMP);
              cachedSize += memoryCacheEntrySize(kv.getBytes(), kv.getBytes(), TOPIC) +
                  8 + // timestamp
                  4; // sequenceNumber
          }
          return i;
      }
  
  }
  