@Test
  public void testBucketingWithNewEntriesDuringScan() throws StoreException, InterruptedException {
    assumeTrue(bucketingEnabled);
    CountDownLatch scanStartedLatch = new CountDownLatch(1);
    CountDownLatch scanHoldLatch = new CountDownLatch(1);
    CountDownLatch secondCheckpointLatch = new CountDownLatch(1);
    CountDownLatch secondCheckpointHoldLatch = new CountDownLatch(1);
    int throttleCountAtInterest = state.referenceIndex.size() + 1;
    throttlers.put(BlobStoreStats.IO_SCHEDULER_JOB_TYPE,
        new MockThrottler(scanStartedLatch, scanHoldLatch, secondCheckpointLatch, secondCheckpointHoldLatch,
            throttleCountAtInterest));
    int bucketCount = 2 * (int) (state.time.milliseconds() / BUCKET_SPAN_IN_MS);
    BlobStoreStats blobStoreStats = setupBlobStoreStats(bucketCount, state.time.milliseconds());
    // proceed only when the scan is started
    assertTrue("IndexScanner took too long to start", scanStartedLatch.await(5, TimeUnit.SECONDS));
    // 1 new put that with expiry
    long expiresAtInMs = ((long) bucketCount - 2) * BUCKET_SPAN_IN_MS;
    List<IndexEntry> newPutEntries = state.addPutEntries(1, PUT_RECORD_SIZE, expiresAtInMs);
    // 6 new puts with no expiry
    newPutEntries.addAll(state.addPutEntries(6, PUT_RECORD_SIZE, Utils.Infinite_Time));
    for (IndexEntry entry : newPutEntries) {
      blobStoreStats.handleNewPutEntry(entry.getKey(), entry.getValue());
    }
    List<MockId> newDeletes = new ArrayList<>();
    // 1 delete from the first index segment
    newDeletes.add(state.getIdToDeleteFromIndexSegment(state.referenceIndex.firstKey(), false));
    // 1 delete from the last index segment
    newDeletes.add(state.getIdToDeleteFromIndexSegment(state.referenceIndex.lastKey(), false));
    for (MockId idToDelete : newDeletes) {
      if (idToDelete != null) {
        newDelete(blobStoreStats, idToDelete);
      }
    }
    // continue the scan towards the first checkpoint
    scanHoldLatch.countDown();
    assertTrue("IndexScanner took too long to start the second checkpoint",
        secondCheckpointLatch.await(5, TimeUnit.SECONDS));
    // add more entries after the second checkpoint is taken
    // 3 new puts with no expiry
    newPutEntries = state.addPutEntries(3, PUT_RECORD_SIZE, Utils.Infinite_Time);
    for (IndexEntry entry : newPutEntries) {
      blobStoreStats.handleNewPutEntry(entry.getKey(), entry.getValue());
    }
    newDeletes.clear();
    // 1 delete from the last index segment
    newDeletes.add(state.getIdToDeleteFromIndexSegment(state.referenceIndex.lastKey(), false));
    for (MockId idToDelete : newDeletes) {
      if (idToDelete != null) {
        newDelete(blobStoreStats, idToDelete);
      }
    }
    // continue the scan towards the second checkpoint
    secondCheckpointHoldLatch.countDown();
    // a probe put with a latch to inform us about the state of the queue
    CountDownLatch queueProcessedLatch = new CountDownLatch(1);
    blobStoreStats.handleNewPutEntry(null, new MockIndexValue(queueProcessedLatch, state.index.getCurrentEndOffset()));
    assertTrue("QueueProcessor took too long to process the new entries",
        queueProcessedLatch.await(3, TimeUnit.SECONDS));
    // verifying log segment from beginning time of the state, since most of the deletes and expiration starting from this time.
    for (long i = state.beginningTime; i <= state.time.milliseconds() + TEST_TIME_INTERVAL_IN_MS;
        i += TEST_TIME_INTERVAL_IN_MS) {
      verifyAndGetLogSegmentValidSize(blobStoreStats, new TimeRange(i, 0L));
    }
    verifyContainerStorageStatsAndGetTotalValidSize(blobStoreStats, state.time.milliseconds());
    long timeToLiveInMs = expiresAtInMs - state.time.milliseconds() < 0 ? 0 : expiresAtInMs - state.time.milliseconds();
    state.advanceTime(timeToLiveInMs + Time.MsPerSec);
    advanceTimeToNextSecond();
    verifyContainerStorageStatsAndGetTotalValidSize(blobStoreStats, state.time.milliseconds());
    blobStoreStats.close();
  }
Test to verify that new entries that got added while scanning are being counted properly. That is, no double
counting or miscounting. The behavior is verified via the following steps:
1. Start the scan.
2. Once the scan is started it will be put on hold and various new entries will be added.
3. Resume the scan and wait for the scan to finish the first check point and start working towards the second
   checkpoint. The scan is held again once it starts working towards the second checkpoint.
5. More new entries will be added.
6. Resume the scan and wait for all added entries to be processed by the queue processor.
4. Perform various checks to verify the reported stats.
@throws StoreException
@throws InterruptedException
ambry-store/src/test/java/com/github/ambry/store/BlobStoreStatsTest.java