private void forwardScan(Offset startOffset, Offset endOffset) throws StoreException {
      logger.trace("Forward scanning from {} to {} by IndexScanner for store {}", startOffset, endOffset, storeId);
      SortedMap<Offset, IndexSegment> tailIndexSegments =
          index.getIndexSegments().subMap(index.getIndexSegments().floorKey(startOffset), endOffset);
      int forwardScanEntryCount = 0;
      for (IndexSegment indexSegment : tailIndexSegments.values()) {
        if (cancelled) {
          return;
        }
        diskIOScheduler.getSlice(BlobStoreStats.IO_SCHEDULER_JOB_TYPE, BlobStoreStats.IO_SCHEDULER_JOB_ID,
            indexSegment.size());
        Iterator<IndexEntry> iterator = indexSegment.iterator();
        Map<StoreKey, IndexValue> seenPuts = new HashMap<>();
        while (iterator.hasNext()) {
          forwardScanEntryCount++;
          IndexEntry entry = iterator.next();
          IndexValue indexValue = entry.getValue();
          if (indexValue.getOffset().compareTo(startOffset) >= 0 && indexValue.getOffset().compareTo(endOffset) < 0) {
            // index value is not yet processed and should be processed
            newScanResults.updateContainerPhysicalStorageUsageAndStoreKey(indexValue.getAccountId(),
                indexValue.getContainerId(), indexValue.getSize(), indexValue.isPut() ? 1 : 0);
            if (!indexValue.isPut()) {
              IndexValue originalPut;
              if (indexValue.getOriginalMessageOffset() == indexValue.getOffset().getOffset()) {
                // update record with no put record due to compaction and legacy bugs
                originalPut = null;
              } else {
                FileSpan searchSpan = new FileSpan(index.getStartOffset(), indexSegment.getStartOffset());
                originalPut = seenPuts.get(entry.getKey());
                if (originalPut == null) {
                  originalPut =
                      index.findKey(entry.getKey(), searchSpan, EnumSet.of(PersistentIndex.IndexEntryType.PUT));
                }
              }
              IndexValue previousValue = null;
              if (originalPut != null) {
                // Find the IndexValue right before current IndexValue, it might be original PUT as well.
                previousValue =
                    index.findKey(entry.getKey(), new FileSpan(originalPut.getOffset(), indexValue.getOffset()),
                        EnumSet.allOf(PersistentIndex.IndexEntryType.class));
              }
              if (originalPut == null) {
                newScanResults.updateLogSegmentBaseBucket(indexValue.getOffset().getName(), indexValue.getSize());
              } else if (indexValue.isDelete()) {
                processNewDelete(newScanResults, entry.getKey(), indexValue, previousValue, originalPut);
              } else if (indexValue.isUndelete()) {
                processNewUndelete(newScanResults, entry.getKey(), indexValue, previousValue, originalPut);
              } else if (indexValue.isTtlUpdate()) {
                processNewTtlUpdate(newScanResults, indexValue, originalPut);
              }
            } else {
              // put record
              processNewPut(newScanResults, indexValue);
            }
          }
          if (indexValue.isPut()) {
            seenPuts.put(entry.getKey(), entry.getValue());
          }
        }
      }
      metrics.statsForwardScanEntryCount.update(forwardScanEntryCount);
    }
Perform a forward scan (older to newest entries) from the given start offset to the given end offset. This
forward scan is used to process any new entries that were added to the index while scanning to the first
checkpoint.
@param startOffset the start offset defining the start of the scan range, inclusive (greater than or equal to)
@param endOffset the end offset defining the end of the scan range, exclusive (strictly less than)
@throws StoreException
ambry-store/src/main/java/com/github/ambry/store/BlobStoreStats.java