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.common.header.internals.RecordHeaders;
  import org.apache.kafka.common.serialization.Serdes;
  import org.apache.kafka.common.utils.Bytes;
  import org.apache.kafka.streams.KeyValue;
  import org.apache.kafka.streams.StreamsConfig;
  import org.apache.kafka.streams.kstream.Windowed;
  import org.apache.kafka.streams.kstream.internals.Change;
  import org.apache.kafka.streams.processor.StateStore;
  import org.apache.kafka.streams.processor.StateStoreContext;
  import org.apache.kafka.streams.processor.api.Record;
  import org.apache.kafka.streams.processor.internals.InternalProcessorContext;
  import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
  import org.apache.kafka.streams.processor.internals.ProcessorStateManager;
  import org.apache.kafka.streams.processor.internals.RecordQueue;
  import org.apache.kafka.streams.state.KeyValueIterator;
  import org.apache.kafka.streams.state.StateSerdes;
  import org.apache.kafka.streams.state.WindowStore;
  import org.apache.kafka.streams.state.WindowStoreIterator;
  import org.apache.kafka.streams.state.internals.MergedSortedCacheWindowStoreKeyValueIterator.StoreKeyToWindowKey;
  import org.apache.kafka.streams.state.internals.MergedSortedCacheWindowStoreKeyValueIterator.WindowKeyToBytes;
  import org.apache.kafka.streams.state.internals.PrefixedWindowKeySchemas.KeyFirstWindowKeySchema;
  import org.apache.kafka.streams.state.internals.PrefixedWindowKeySchemas.TimeFirstWindowKeySchema;
  import org.apache.kafka.streams.state.internals.SegmentedBytesStore.KeySchema;
  import org.apache.kafka.streams.state.internals.ThreadCache.DirtyEntry;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.util.HashSet;
  import java.util.LinkedList;
  import java.util.List;
  import java.util.NoSuchElementException;
  import java.util.Set;
  import java.util.concurrent.atomic.AtomicLong;
  import java.util.function.Function;
  
  import static org.apache.kafka.streams.processor.internals.ProcessorContextUtils.asInternalProcessorContext;
  import static org.apache.kafka.streams.state.internals.ExceptionUtils.executeAll;
  import static org.apache.kafka.streams.state.internals.ExceptionUtils.throwSuppressed;
  
  class TimeOrderedCachingWindowStore
      extends WrappedStateStore<WindowStore<Bytes, byte[]>, byte[], byte[]>
      implements WindowStore<Bytes, byte[]>, CachedStateStore<byte[], byte[]> {
  
      private static final Logger LOG = LoggerFactory.getLogger(TimeOrderedCachingWindowStore.class);
  
      private final long windowSize;
      private final SegmentedCacheFunction baseKeyCacheFunction;
      private final SegmentedCacheFunction indexKeyCacheFunction;
      private final TimeFirstWindowKeySchema baseKeySchema = new TimeFirstWindowKeySchema();
      private final KeyFirstWindowKeySchema indexKeySchema = new KeyFirstWindowKeySchema();
  
      private String cacheName;
      private boolean hasIndex;
      private boolean sendOldValues;
      private InternalProcessorContext<?, ?> internalContext;
      private StateSerdes<Bytes, byte[]> bytesSerdes;
      private CacheFlushListener<byte[], byte[]> flushListener;
  
      private final AtomicLong maxObservedTimestamp;
  
      TimeOrderedCachingWindowStore(final WindowStore<Bytes, byte[]> underlying,
                                    final long windowSize,
                                    final long segmentInterval) {
          super(underlying);
          this.windowSize = windowSize;
          this.baseKeyCacheFunction = new SegmentedCacheFunction(baseKeySchema, segmentInterval);
          this.indexKeyCacheFunction = new SegmentedCacheFunction(indexKeySchema, segmentInterval);
          this.maxObservedTimestamp = new AtomicLong(RecordQueue.UNKNOWN);
          enforceWrappedStore(underlying);
      }
  
      private void enforceWrappedStore(final WindowStore<Bytes, byte[]> underlying) {
          final RocksDBTimeOrderedWindowStore timeOrderedWindowStore = getWrappedStore(underlying);
          if (timeOrderedWindowStore == null) {
              throw new IllegalArgumentException("TimeOrderedCachingWindowStore only supports RocksDBTimeOrderedWindowStore backed store");
          }
  
          hasIndex = timeOrderedWindowStore.hasIndex();
      }
  
      @SuppressWarnings("unchecked")
      private RocksDBTimeOrderedWindowStore getWrappedStore(final StateStore wrapped) {
          if (wrapped instanceof RocksDBTimeOrderedWindowStore) {
              return (RocksDBTimeOrderedWindowStore) wrapped;
          }
          if (wrapped instanceof WrappedStateStore) {
              return getWrappedStore(((WrappedStateStore<?, Bytes, byte[]>) wrapped).wrapped());
          }
          return null;
      }
  
      @Override
      public void init(final StateStoreContext stateStoreContext, final StateStore root) {
          final String prefix = StreamsConfig.InternalConfig.getString(
              stateStoreContext.appConfigs(),
              StreamsConfig.InternalConfig.TOPIC_PREFIX_ALTERNATIVE,
              stateStoreContext.applicationId()
          );
          internalContext = asInternalProcessorContext(stateStoreContext);
          final String topic = ProcessorStateManager.storeChangelogTopic(prefix, name(),  stateStoreContext.taskId().topologyName());
  
          bytesSerdes = new StateSerdes<>(
              topic,
              Serdes.Bytes(),
              Serdes.ByteArray());
          cacheName = stateStoreContext.taskId() + "-" + name();
  
          internalContext.registerCacheFlushListener(cacheName, entries -> putAndMaybeForward(entries, internalContext));
          super.init(stateStoreContext, root);
      }
  
      private void putAndMaybeForward(final List<DirtyEntry> entries,
                                      final InternalProcessorContext<?, ?> context) {
  
          // Track what base key or index key we already processed so don't reprocess
          final Set<Bytes> processedBasedKey = new HashSet<>();
  
          for (final ThreadCache.DirtyEntry entry : entries) {
              final byte[] binaryWindowKey = baseKeyCacheFunction.key(entry.key()).get();
              final boolean isBaseKey = PrefixedWindowKeySchemas.isTimeFirstSchemaKey(
                  binaryWindowKey);
  
              final DirtyEntry finalEntry;
              if (!isBaseKey) {
                  final Bytes baseKey = indexKeyToBaseKey(Bytes.wrap(binaryWindowKey));
                  if (hasIndex && processedBasedKey.contains(baseKey)) {
                      // Processed in base
                      continue;
                  }
  
                  final Bytes cachedBaseKey = baseKeyCacheFunction.cacheKey(baseKey);
                  final LRUCacheEntry value = context.cache().get(cacheName, cachedBaseKey);
                  // Base key value is already evicted, which should be handled already
                  if (value == null) {
                      continue;
                  }
  
                  finalEntry = new DirtyEntry(entry.key(), value.value(), value);
  
                  if (hasIndex) {
                      processedBasedKey.add(baseKey);
                  }
              } else {
                  final Bytes baseKey = Bytes.wrap(binaryWindowKey);
                  if (hasIndex && processedBasedKey.contains(baseKey)) {
                      // Processed in index
                      continue;
                  }
                  finalEntry = entry;
                  if (hasIndex) {
                      processedBasedKey.add(Bytes.wrap(binaryWindowKey));
                  }
              }
  
              final Windowed<Bytes> windowedKeyBytes;
              if (isBaseKey) {
                  windowedKeyBytes = TimeFirstWindowKeySchema.fromStoreBytesKey(binaryWindowKey,
                      windowSize);
              } else {
                  windowedKeyBytes = KeyFirstWindowKeySchema.fromStoreBytesKey(binaryWindowKey,
                      windowSize);
              }
  
              final long windowStartTimestamp = windowedKeyBytes.window().start();
              final Bytes binaryKey = windowedKeyBytes.key();
  
              putAndMaybeForward(context, finalEntry, binaryKey, windowStartTimestamp);
          }
      }
  
      private void putAndMaybeForward(final InternalProcessorContext<?, ?> context,
                                      final DirtyEntry finalEntry,
                                      final Bytes binaryKey,
                                      final long windowStartTimestamp) {
          if (flushListener != null) {
              final byte[] rawNewValue = finalEntry.newValue();
              final byte[] rawOldValue = rawNewValue == null || sendOldValues ?
                  wrapped().fetch(binaryKey, windowStartTimestamp) : null;
  
              // this is an optimization: if this key did not exist in underlying store and also not in the cache,
              // we can skip flushing to downstream as well as writing to underlying store
              if (rawNewValue != null || rawOldValue != null) {
                  // we need to get the old values if needed, and then put to store, and then flush
                  final ProcessorRecordContext current = context.recordContext();
                  try {
                      context.setRecordContext(finalEntry.entry().context());
                      wrapped().put(binaryKey, finalEntry.newValue(), windowStartTimestamp);
  
                      flushListener.apply(
                          new Record<>(
                              WindowKeySchema.toStoreKeyBinary(binaryKey,
                                      windowStartTimestamp, 0)
                                  .get(),
                              new Change<>(rawNewValue, sendOldValues ? rawOldValue : null),
                              finalEntry.entry().context().timestamp(),
                              finalEntry.entry().context().headers()));
                  } finally {
                      context.setRecordContext(current);
                  }
              }
          } else {
              final ProcessorRecordContext current = context.recordContext();
              try {
                  context.setRecordContext(finalEntry.entry().context());
                  wrapped().put(binaryKey, finalEntry.newValue(), windowStartTimestamp);
              } finally {
                  context.setRecordContext(current);
              }
          }
      }
  
      @Override
      public boolean setFlushListener(final CacheFlushListener<byte[], byte[]> flushListener,
                                      final boolean sendOldValues) {
          this.flushListener = flushListener;
          this.sendOldValues = sendOldValues;
  
          return true;
      }
  
      private Bytes indexKeyToBaseKey(final Bytes indexKey) {
          final byte[] key = KeyFirstWindowKeySchema.extractStoreKeyBytes(indexKey.get());
          final long timestamp = KeyFirstWindowKeySchema.extractStoreTimestamp(indexKey.get());
          final int seqnum = KeyFirstWindowKeySchema.extractStoreSequence(indexKey.get());
          return TimeFirstWindowKeySchema.toStoreKeyBinary(key, timestamp, seqnum);
      }
  
      @Override
      public synchronized void put(final Bytes key,
                                   final byte[] value,
                                   final long windowStartTimestamp) {
          // since this function may not access the underlying inner store, we need to validate
          // if store is open outside as well.
          validateStoreOpen();
  
          final Bytes baseKeyBytes = TimeFirstWindowKeySchema.toStoreKeyBinary(key, windowStartTimestamp, 0);
          final LRUCacheEntry entry =
              new LRUCacheEntry(
                  value,
                  internalContext.recordContext().headers(),
                  true,
                  internalContext.recordContext().offset(),
                  internalContext.recordContext().timestamp(),
                  internalContext.recordContext().partition(),
                  internalContext.recordContext().topic(),
                  internalContext.recordContext().sourceRawKey(),
                  internalContext.recordContext().sourceRawValue()
              );
  
          // Put to index first so that base can be evicted later
          if (hasIndex) {
              // Important: put base key first to avoid the situation that if we put index first,
              // it could be evicted when we are putting base key. In that case, base key is not yet
              // in cache so we can't store key/value to store when index is evicted. Then if we fetch
              // using index, we can't find it in either store or cache
              internalContext.cache().put(cacheName, baseKeyCacheFunction.cacheKey(baseKeyBytes), entry);
              final LRUCacheEntry emptyEntry =
                  new LRUCacheEntry(
                      new byte[0],
                      new RecordHeaders(),
                      true,
                      internalContext.recordContext().offset(),
                      internalContext.recordContext().timestamp(),
                      internalContext.recordContext().partition(),
                      "",
                      internalContext.recordContext().sourceRawKey(),
                      internalContext.recordContext().sourceRawValue()
                  );
              final Bytes indexKey = KeyFirstWindowKeySchema.toStoreKeyBinary(key, windowStartTimestamp, 0);
              internalContext.cache().put(cacheName, indexKeyCacheFunction.cacheKey(indexKey), emptyEntry);
          } else {
              internalContext.cache().put(cacheName, baseKeyCacheFunction.cacheKey(baseKeyBytes), entry);
          }
          maxObservedTimestamp.set(Math.max(windowStartTimestamp, maxObservedTimestamp.get()));
      }
  
      @Override
      public byte[] fetch(final Bytes key,
                          final long timestamp) {
          validateStoreOpen();
          if (internalContext.cache() == null) {
              return wrapped().fetch(key, timestamp);
          }
  
          final Bytes baseBytesKey = TimeFirstWindowKeySchema.toStoreKeyBinary(key, timestamp, 0);
          final Bytes cacheKey = baseKeyCacheFunction.cacheKey(baseBytesKey);
  
          final LRUCacheEntry entry = internalContext.cache().get(cacheName, cacheKey);
          if (entry == null) {
              return wrapped().fetch(key, timestamp);
          } else {
              return entry.value();
          }
      }
  
      @Override
      public synchronized WindowStoreIterator<byte[]> fetch(final Bytes key,
                                                            final long timeFrom,
                                                            final long timeTo) {
          // since this function may not access the underlying inner store, we need to validate
          // if store is open outside as well.
          validateStoreOpen();
  
          final WindowStoreIterator<byte[]> underlyingIterator = wrapped().fetch(key, timeFrom, timeTo);
          if (internalContext.cache() == null) {
              return underlyingIterator;
          }
  
          return fetchInternal(underlyingIterator, key, timeFrom, timeTo, true);
      }
  
      @Override
      public synchronized WindowStoreIterator<byte[]> backwardFetch(final Bytes key,
                                                                    final long timeFrom,
                                                                    final long timeTo) {
          // since this function may not access the underlying inner store, we need to validate
          // if store is open outside as well.
          validateStoreOpen();
  
          final WindowStoreIterator<byte[]> underlyingIterator = wrapped().backwardFetch(key, timeFrom, timeTo);
          if (internalContext.cache() == null) {
              return underlyingIterator;
          }
  
          return fetchInternal(underlyingIterator, key, timeFrom, timeTo, false);
      }
  
      private WindowStoreIterator<byte[]> fetchInternal(final WindowStoreIterator<byte[]> underlyingIterator,
                                                        final Bytes key,
                                                        final long timeFrom,
                                                        final long timeTo,
                                                        final boolean forward) {
          final PeekingKeyValueIterator<Bytes, LRUCacheEntry> cacheIterator = new CacheIteratorWrapper(
              key, timeFrom, timeTo, forward, hasIndex);
          final KeySchema keySchema = hasIndex ? indexKeySchema : baseKeySchema;
          final SegmentedCacheFunction cacheFunction = hasIndex ? indexKeyCacheFunction : baseKeyCacheFunction;
          final HasNextCondition hasNextCondition = keySchema.hasNextCondition(key, key, timeFrom, timeTo, forward);
  
          final PeekingKeyValueIterator<Bytes, LRUCacheEntry> filteredCacheIterator =
              new FilteredCacheIterator(cacheIterator, hasNextCondition, cacheFunction);
  
          final Function<byte[], Long> tsExtractor = hasIndex ? KeyFirstWindowKeySchema::extractStoreTimestamp
              : TimeFirstWindowKeySchema::extractStoreTimestamp;
          return new MergedSortedCacheWindowStoreIterator(filteredCacheIterator, underlyingIterator, forward, tsExtractor);
      }
  
      @Override
      public KeyValueIterator<Windowed<Bytes>, byte[]> fetch(final Bytes keyFrom,
                                                             final Bytes keyTo,
                                                             final long timeFrom,
                                                             final long timeTo) {
          if (keyFrom != null && keyTo != null && keyFrom.compareTo(keyTo) > 0) {
              LOG.warn("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");
              return KeyValueIterators.emptyIterator();
          }
  
          // since this function may not access the underlying inner store, we need to validate
          // if store is open outside as well.
          validateStoreOpen();
  
          final KeyValueIterator<Windowed<Bytes>, byte[]> underlyingIterator =
              wrapped().fetch(keyFrom, keyTo, timeFrom, timeTo);
          if (internalContext.cache() == null) {
              return underlyingIterator;
          }
  
          return fetchKeyRange(underlyingIterator, keyFrom, keyTo, timeFrom, timeTo, true);
      }
  
      @Override
      public KeyValueIterator<Windowed<Bytes>, byte[]> backwardFetch(final Bytes keyFrom,
                                                                     final Bytes keyTo,
                                                                     final long timeFrom,
                                                                     final long timeTo) {
          if (keyFrom != null && keyTo != null && keyFrom.compareTo(keyTo) > 0) {
              LOG.warn("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");
              return KeyValueIterators.emptyIterator();
          }
  
          // since this function may not access the underlying inner store, we need to validate
          // if store is open outside as well.
          validateStoreOpen();
  
          final KeyValueIterator<Windowed<Bytes>, byte[]> underlyingIterator =
              wrapped().backwardFetch(keyFrom, keyTo, timeFrom, timeTo);
          if (internalContext.cache() == null) {
              return underlyingIterator;
          }
  
          return fetchKeyRange(underlyingIterator, keyFrom, keyTo, timeFrom, timeTo, false);
      }
  
      private KeyValueIterator<Windowed<Bytes>, byte[]> fetchKeyRange(final KeyValueIterator<Windowed<Bytes>, byte[]> underlyingIterator,
                                                                      final Bytes keyFrom,
                                                                      final Bytes keyTo,
                                                                      final long timeFrom,
                                                                      final long timeTo,
                                                                      final boolean forward) {
          final PeekingKeyValueIterator<Bytes, LRUCacheEntry> cacheIterator = new CacheIteratorWrapper(
              keyFrom, keyTo, timeFrom, timeTo, forward, hasIndex);
  
          final KeySchema keySchema = hasIndex ? indexKeySchema : baseKeySchema;
          final HasNextCondition hasNextCondition = keySchema.hasNextCondition(keyFrom, keyTo, timeFrom, timeTo, forward);
          final SegmentedCacheFunction cacheFunction = hasIndex ? indexKeyCacheFunction : baseKeyCacheFunction;
  
          final PeekingKeyValueIterator<Bytes, LRUCacheEntry> filteredCacheIterator =
              new FilteredCacheIterator(cacheIterator, hasNextCondition, cacheFunction);
          final StoreKeyToWindowKey storeKeyToWindowKey = hasIndex ? KeyFirstWindowKeySchema::fromStoreKey : TimeFirstWindowKeySchema::fromStoreKey;
          final WindowKeyToBytes windowKeyToBytes = hasIndex ? KeyFirstWindowKeySchema::toStoreKeyBinary : TimeFirstWindowKeySchema::toStoreKeyBinary;
  
          return new MergedSortedCacheWindowStoreKeyValueIterator(
              filteredCacheIterator,
              underlyingIterator,
              bytesSerdes,
              windowSize,
              cacheFunction,
              forward,
              storeKeyToWindowKey,
              windowKeyToBytes
          );
      }
  
      @Override
      public KeyValueIterator<Windowed<Bytes>, byte[]> fetchAll(final long timeFrom,
                                                                final long timeTo) {
          validateStoreOpen();
  
          final KeyValueIterator<Windowed<Bytes>, byte[]> underlyingIterator = wrapped().fetchAll(timeFrom, timeTo);
          return fetchAllInternal(underlyingIterator, timeFrom, timeTo, true);
      }
  
      @Override
      public KeyValueIterator<Windowed<Bytes>, byte[]> backwardFetchAll(final long timeFrom,
                                                                        final long timeTo) {
          validateStoreOpen();
  
          final KeyValueIterator<Windowed<Bytes>, byte[]> underlyingIterator = wrapped().backwardFetchAll(timeFrom, timeTo);
          return fetchAllInternal(underlyingIterator, timeFrom, timeTo, false);
      }
  
      private KeyValueIterator<Windowed<Bytes>, byte[]> fetchAllInternal(final KeyValueIterator<Windowed<Bytes>, byte[]> underlyingIterator,
                                                                         final long timeFrom,
                                                                         final long timeTo,
                                                                         final boolean forward) {
          final PeekingKeyValueIterator<Bytes, LRUCacheEntry> cacheIterator = new CacheIteratorWrapper(
              null, null, timeFrom, timeTo, forward, false);
          final HasNextCondition hasNextCondition = baseKeySchema.hasNextCondition(null, null, timeFrom, timeTo, forward);
          final PeekingKeyValueIterator<Bytes, LRUCacheEntry> filteredCacheIterator =
              new FilteredCacheIterator(cacheIterator, hasNextCondition, baseKeyCacheFunction);
  
          final StoreKeyToWindowKey storeKeyToWindowKey = TimeFirstWindowKeySchema::fromStoreKey;
          final WindowKeyToBytes windowKeyToBytes = TimeFirstWindowKeySchema::toStoreKeyBinary;
  
          return new MergedSortedCacheWindowStoreKeyValueIterator(
              filteredCacheIterator,
              underlyingIterator,
              bytesSerdes,
              windowSize,
              baseKeyCacheFunction,
              forward,
              storeKeyToWindowKey,
              windowKeyToBytes
          );
      }
  
      @Override
      public KeyValueIterator<Windowed<Bytes>, byte[]> all() {
          validateStoreOpen();
  
          final KeyValueIterator<Windowed<Bytes>, byte[]> underlyingIterator = wrapped().all();
          return fetchAllInternal(underlyingIterator, 0, Long.MAX_VALUE, true);
      }
  
      @Override
      public KeyValueIterator<Windowed<Bytes>, byte[]> backwardAll() {
          validateStoreOpen();
  
          final KeyValueIterator<Windowed<Bytes>, byte[]> underlyingIterator = wrapped().backwardAll();
          return fetchAllInternal(underlyingIterator, 0, Long.MAX_VALUE, false);
      }
  
      @Override
      public synchronized void flush() {
          internalContext.cache().flush(cacheName);
          wrapped().flush();
      }
  
      @Override
      public void flushCache() {
          internalContext.cache().flush(cacheName);
      }
  
      @Override
      public void clearCache() {
          internalContext.cache().clear(cacheName);
      }
  
      @Override
      public synchronized void close() {
          final LinkedList<RuntimeException> suppressed = executeAll(
              () -> internalContext.cache().flush(cacheName),
              () -> internalContext.cache().close(cacheName),
              wrapped()::close
          );
          if (!suppressed.isEmpty()) {
              throwSuppressed("Caught an exception while closing caching window store for store " + name(),
                  suppressed);
          }
      }
  
  
      private class CacheIteratorWrapper implements PeekingKeyValueIterator<Bytes, LRUCacheEntry> {
  
          private final long segmentInterval;
          private final Bytes keyFrom;
          private final Bytes keyTo;
          private final long timeTo;
          private final boolean forward;
          private final boolean useIndex; // If we are iterating from index
  
          private long lastSegmentId;
          private long currentSegmentId;
          private Bytes cacheKeyFrom;
          private Bytes cacheKeyTo;
          private LRUCacheEntry cachedBaseValue;
          private final SegmentedCacheFunction cacheFunction;
  
          private ThreadCache.MemoryLRUCacheBytesIterator current;
  
          private CacheIteratorWrapper(final Bytes key,
                                       final long timeFrom,
                                       final long timeTo,
                                       final boolean forward,
                                       final boolean index) {
              this(key, key, timeFrom, timeTo, forward, index);
          }
  
          private CacheIteratorWrapper(final Bytes keyFrom,
                                       final Bytes keyTo,
                                       final long timeFrom,
                                       final long timeTo,
                                       final boolean forward,
                                       final boolean index) {
              this.keyFrom = keyFrom;
              this.keyTo = keyTo;
              this.timeTo = timeTo;
              this.forward = forward;
              this.useIndex = index;
  
              cacheFunction = index ? indexKeyCacheFunction : baseKeyCacheFunction;
  
              this.segmentInterval = cacheFunction.getSegmentInterval();
  
              if (forward) {
                  this.lastSegmentId = cacheFunction.segmentId(Math.min(timeTo, maxObservedTimestamp.get()));
                  this.currentSegmentId = cacheFunction.segmentId(timeFrom);
  
                  setCacheKeyRange(timeFrom, currentSegmentLastTime());
                  this.current = internalContext.cache().range(cacheName, cacheKeyFrom, cacheKeyTo);
              } else {
                  this.currentSegmentId = cacheFunction.segmentId(Math.min(timeTo, maxObservedTimestamp.get()));
                  this.lastSegmentId = cacheFunction.segmentId(timeFrom);
  
                  setCacheKeyRange(currentSegmentBeginTime(), Math.min(timeTo, maxObservedTimestamp.get()));
                  this.current = internalContext.cache().reverseRange(cacheName, cacheKeyFrom, cacheKeyTo);
              }
          }
  
          @Override
          public boolean hasNext() {
              if (current == null) {
                  return false;
              }
  
              if (useIndex) {
                  do {
                      // If iterating from index, need to make sure base key/value exist in cache
                      while (current.hasNext()) {
                          final Bytes cacheIndexKey = current.peekNextKey();
                          final Bytes indexKey = indexKeyCacheFunction.key(cacheIndexKey);
                          final Bytes baseKey = indexKeyToBaseKey(indexKey);
                          final Bytes cachedBaseKey = baseKeyCacheFunction.cacheKey(baseKey);
                          cachedBaseValue = internalContext.cache().get(cacheName, cachedBaseKey);
                          if (cachedBaseValue != null) {
                              return true;
                          }
                          current.next();
                      }
                      getNextSegmentIterator();
                  } while (current != null);
                  return false;
              }
  
              if (current.hasNext()) {
                  return true;
              }
  
              while (!current.hasNext()) {
                  getNextSegmentIterator();
                  if (current == null) {
                      return false;
                  }
              }
              return true;
          }
  
          @Override
          public Bytes peekNextKey() {
              if (!hasNext()) {
                  throw new NoSuchElementException();
              }
              return current.peekNextKey();
          }
  
          @Override
          public KeyValue<Bytes, LRUCacheEntry> peekNext() {
              if (!hasNext()) {
                  throw new NoSuchElementException();
              }
              if (useIndex) {
                  final KeyValue<Bytes, LRUCacheEntry> kv = current.peekNext();
                  return KeyValue.pair(kv.key, cachedBaseValue);
              }
              return current.peekNext();
          }
  
          @Override
          public KeyValue<Bytes, LRUCacheEntry> next() {
              if (!hasNext()) {
                  throw new NoSuchElementException();
              }
              if (useIndex) {
                  final KeyValue<Bytes, LRUCacheEntry> kv = current.next();
                  return KeyValue.pair(kv.key, cachedBaseValue);
              }
              return current.next();
          }
  
          @Override
          public void close() {
              current.close();
          }
  
          private long currentSegmentBeginTime() {
              return currentSegmentId * segmentInterval;
          }
  
          private long currentSegmentLastTime() {
              return Math.min(timeTo, currentSegmentBeginTime() + segmentInterval - 1);
          }
  
          private void getNextSegmentIterator() {
              if (forward) {
                  ++currentSegmentId;
                  // updating as maxObservedTimestamp can change while iterating
                  lastSegmentId = cacheFunction.segmentId(Math.min(timeTo, maxObservedTimestamp.get()));
  
                  if (currentSegmentId > lastSegmentId) {
                      current = null;
                      return;
                  }
  
                  setCacheKeyRange(currentSegmentBeginTime(), currentSegmentLastTime());
  
                  current.close();
  
                  current = internalContext.cache().range(cacheName, cacheKeyFrom, cacheKeyTo);
              } else {
                  --currentSegmentId;
  
                  // last segment id is stable when iterating backward, therefore no need to update
                  if (currentSegmentId < lastSegmentId) {
                      current = null;
                      return;
                  }
  
                  setCacheKeyRange(currentSegmentBeginTime(), currentSegmentLastTime());
  
                  current.close();
  
                  current = internalContext.cache().reverseRange(cacheName, cacheKeyFrom, cacheKeyTo);
              }
          }
  
          private void setCacheKeyRange(final long lowerRangeEndTime, final long upperRangeEndTime) {
              if (cacheFunction.segmentId(lowerRangeEndTime) != cacheFunction.segmentId(upperRangeEndTime)) {
                  throw new IllegalStateException("Error iterating over segments: segment interval has changed");
              }
  
              final KeySchema schema = useIndex ? indexKeySchema : baseKeySchema;
  
              if (keyFrom != null && keyFrom.equals(keyTo)) {
                  cacheKeyFrom = cacheFunction.cacheKey(schema.lowerRangeFixedSize(keyFrom, lowerRangeEndTime), currentSegmentId);
                  cacheKeyTo = cacheFunction.cacheKey(schema.upperRangeFixedSize(keyTo, upperRangeEndTime), currentSegmentId);
              } else {
                  cacheKeyFrom = cacheFunction.cacheKey(schema.lowerRange(keyFrom, lowerRangeEndTime), currentSegmentId);
                  cacheKeyTo = cacheFunction.cacheKey(schema.upperRange(keyTo, timeTo), currentSegmentId);
              }
          }
      }
  }
  