@Override
    public void run() {
      try {
        int entryCount;
        ScanResults currentScanResults = scanResults.get();
        // prevent new index entries getting dropped when IndexScanner is started before the QueueProcessor while
        // new index entries are being added. e.g. IndexScanner started and recorded the end offset as 100, meanwhile
        // new entries are added and IndexScanner is not finished yet. Before IndexScanner can finish, QueueProcessor
        // is started and it's going to process new entries with offset 90 to 120. Since IndexScanner is still in
        // progress the QueueProcessor will operate on the old ScanResults and index entries between 100 and
        // 120 will be missing for the next forecast period.
        if (isScanning || currentScanResults == null) {
          return;
        } else {
          entryCount = queueEntryCount.get();
        }
        long processStartTimeMs = time.milliseconds();
        for (int i = 0; i < entryCount && !cancelled && !isScanning; i++) {
          EntryContext entryContext = recentEntryQueue.poll();
          if (entryContext == null) {
            throw new IllegalStateException(
                String.format("Invalid queue state in store %s. Expected entryCount %d, current index: %d", storeId,
                    entryCount, i));
          }
          metrics.statsRecentEntryQueueSize.update(queueEntryCount.decrementAndGet());
          StoreKey key = entryContext.key;
          IndexValue newValue = entryContext.currentValue;
          IndexValue originalPut = entryContext.originalPutValue;
          IndexValue previousValue = entryContext.previousValue;
          // prevent double counting new entries that were added after enabling the queue and just before the second
          // checkpoint is taken
          if (newValue.getOffset().compareTo(currentScanResults.scannedEndOffset) >= 0) {
            currentScanResults.updateContainerPhysicalStorageUsageAndStoreKey(newValue.getAccountId(),
                newValue.getContainerId(), newValue.getSize(), newValue.isPut() ? 1 : 0);
            if (newValue.isDelete()) {
              // new delete
              processNewDelete(currentScanResults, key, newValue, originalPut, previousValue);
            } else if (newValue.isUndelete()) {
              // new undelete
              processNewUndelete(currentScanResults, key, newValue, originalPut, previousValue);
            } else if (newValue.isTtlUpdate()) {
              // new ttl update
              processNewTtlUpdate(currentScanResults, newValue, originalPut);
            } else {
              // new put
              processNewPut(currentScanResults, newValue);
            }
          }
        }
        metrics.statsRecentEntryQueueProcessTimeMs.update(time.milliseconds() - processStartTimeMs,
            TimeUnit.MILLISECONDS);
      } catch (Exception e) {
        logger.error("Unexpected exception while running QueueProcessor in store {}", storeId, e);
        metrics.blobStoreStatsQueueProcessorErrorCount.inc();
      }
    }
Runner that processes new entries buffered in the recentEntryQueue and make appropriate updates to keep the current
{@link ScanResults} relevant.
ambry-store/src/main/java/com/github/ambry/store/BlobStoreStats.java