@Test
  public void testBucketingWithNewEntriesAfterScan() throws StoreException, InterruptedException {
    assumeTrue(bucketingEnabled);
    CountDownLatch scanStartedLatch = new CountDownLatch(1);
    CountDownLatch queueProcessedLatch = new CountDownLatch(1);
    MockThrottler mockThrottler = new MockThrottler(scanStartedLatch, new CountDownLatch(0));
    throttlers.put(BlobStoreStats.IO_SCHEDULER_JOB_TYPE, mockThrottler);
    // add a put that will expire immediately
    state.addPutEntries(1, PUT_RECORD_SIZE, state.time.milliseconds());
    advanceTimeToNextSecond();
    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));
    advanceTimeToNextSecond();
    verifyAndGetLogSegmentValidSize(blobStoreStats, new TimeRange(state.time.milliseconds(), 0L));
    verifyContainerStorageStatsAndGetTotalValidSize(blobStoreStats, state.time.milliseconds());
    int throttleCountBeforeRequests = mockThrottler.throttleCount.get();
    // add 3 put that are expiring within forecast range
    long expiresAtInMs = ((long) bucketCount - 2) * BUCKET_SPAN_IN_MS;
    List<IndexEntry> newPutEntries = state.addPutEntries(3, PUT_RECORD_SIZE, expiresAtInMs);
    // add a put that will expire outside of the forecast coverage
    newPutEntries.addAll(state.addPutEntries(1, PUT_RECORD_SIZE, ((long) bucketCount) * BUCKET_SPAN_IN_MS));
    for (IndexEntry entry : newPutEntries) {
      blobStoreStats.handleNewPutEntry(entry.getKey(), entry.getValue());
    }
    // add a long expired put
    List<IndexEntry> expiredPutEntries = state.addPutEntries(1, PUT_RECORD_SIZE, 0);
    // add a put that will expire immediately
    expiredPutEntries.addAll(state.addPutEntries(1, PUT_RECORD_SIZE, state.time.milliseconds()));
    for (IndexEntry entry : expiredPutEntries) {
      blobStoreStats.handleNewPutEntry(entry.getKey(), entry.getValue());
    }
    advanceTimeToNextSecond();
    // delete the first new put that is expiring
    MockId firstPutWithExpiry = getIdToDelete(newPutEntries.get(0).getKey());
    newDelete(blobStoreStats, firstPutWithExpiry);
    // delete the puts that are already expired
    for (IndexEntry entry : expiredPutEntries) {
      newDelete(blobStoreStats, (MockId) entry.getKey());
    }
    // add two blobs that will expired fairly soon, testing TTL_UPDATE and DELETE
    long soonExpiredTimeInMs = state.time.milliseconds() + (10 * BUCKET_SPAN_IN_MS);
    List<IndexEntry> soonExpiredPutEntries = state.addPutEntries(2, PUT_RECORD_SIZE, soonExpiredTimeInMs);
    for (IndexEntry entry : soonExpiredPutEntries) {
      blobStoreStats.handleNewPutEntry(entry.getKey(), entry.getValue());
      newTtlUpdate(blobStoreStats, (MockId) entry.getKey());
    }
    // Delete first entry
    newDelete(blobStoreStats, getIdToDelete(soonExpiredPutEntries.get(0).getKey()));
    advanceTimeToNextSecond();
    newDelete(blobStoreStats, (MockId) soonExpiredPutEntries.get(0).getKey(), (short) 2);
    advanceTimeToNextSecond();

    // Testing UNDELETE and DELETE
    // P, D, U
    // P, U
    // P, T, D, U
    // P, D, U, T
    // P, T, D, U, U
    // P, D, D, U
    // P, D, U, D, U
    long expiredForUndeleteEntries = state.time.milliseconds() + 100 * BUCKET_SPAN_IN_MS;
    List<IndexEntry> toUndeleteEntries = state.addPutEntries(7, PUT_RECORD_SIZE, expiredForUndeleteEntries);
    for (IndexEntry entry : toUndeleteEntries) {
      blobStoreStats.handleNewPutEntry(entry.getKey(), entry.getValue());
    }
    MockId id1 = getIdToDelete(toUndeleteEntries.get(0).getKey());
    newDelete(blobStoreStats, id1);
    newUndelete(blobStoreStats, id1);

    MockId id2 = (MockId) toUndeleteEntries.get(1).getKey();
    newUndelete(blobStoreStats, id2, (short) 1);

    MockId id3 = (MockId) toUndeleteEntries.get(2).getKey();
    newTtlUpdate(blobStoreStats, id3);
    newDelete(blobStoreStats, getIdToDelete(id3));
    newUndelete(blobStoreStats, id3);

    MockId id4 = (MockId) toUndeleteEntries.get(3).getKey();
    newDelete(blobStoreStats, getIdToDelete(id4));
    newUndelete(blobStoreStats, id4);
    newTtlUpdate(blobStoreStats, id4);

    MockId id5 = (MockId) toUndeleteEntries.get(4).getKey();
    newTtlUpdate(blobStoreStats, id5);
    newDelete(blobStoreStats, getIdToDelete(id5));
    newUndelete(blobStoreStats, id5);
    newUndelete(blobStoreStats, id5, (short) 2);

    MockId id6 = (MockId) toUndeleteEntries.get(5).getKey();
    newDelete(blobStoreStats, getIdToDelete(id6));
    advanceTimeToNextSecond();
    newDelete(blobStoreStats, id6, (short) 1);
    newUndelete(blobStoreStats, id6);
    advanceTimeToNextSecond();

    MockId id7 = (MockId) toUndeleteEntries.get(6).getKey();
    newDelete(blobStoreStats, getIdToDelete(id7));
    newUndelete(blobStoreStats, id7);
    newDelete(blobStoreStats, id7);
    newUndelete(blobStoreStats, id7);

    // a probe put with a latch to inform us about the state of the queue
    blobStoreStats.handleNewPutEntry(null, new MockIndexValue(queueProcessedLatch, state.index.getCurrentEndOffset()));
    assertTrue("QueueProcessor took too long to process the new entries",
        queueProcessedLatch.await(Long.MAX_VALUE, TimeUnit.SECONDS));
    for (long i = 0; 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());
    // add 3 put with no expiry
    newPutEntries = state.addPutEntries(3, PUT_RECORD_SIZE, Utils.Infinite_Time);
    for (IndexEntry entry : newPutEntries) {
      blobStoreStats.handleNewPutEntry(entry.getKey(), entry.getValue());
    }
    // delete one of the newly added put
    newDelete(blobStoreStats, state.getIdToDeleteFromIndexSegment(state.referenceIndex.lastKey(), false));
    queueProcessedLatch = new CountDownLatch(1);
    // a probe put with a latch to inform us about the state of the queue
    blobStoreStats.handleNewPutEntry(null, new MockIndexValue(queueProcessedLatch, state.index.getCurrentEndOffset()));
    assertTrue("QueueProcessor took too long to process the new entries",
        queueProcessedLatch.await(Long.MAX_VALUE, TimeUnit.SECONDS));
    // note: advance time only after all new deletes are added to avoid the scenario where an index segment's last
    // modified time changes after being processed by BlobStoreStats. E.g. if the following operations are made in
    // order: new put, new delete, advance time, new put, etc..., then there is a chance where BlobStoreStats processed
    // the new delete with t1 but CuratedLogIndexState could bump the last modified time to t2 (if the new put after
    // advancing the time is still in the same index segment as the delete).
    advanceTimeToNextSecond();
    for (long i = 0; 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());
    // advance time beyond expiration of the blobs and verify no double counting for expiration and delete
    long timeToLiveInMs = expiresAtInMs - state.time.milliseconds() < 0 ? 0 : expiresAtInMs - state.time.milliseconds();
    state.advanceTime(timeToLiveInMs + Time.MsPerSec);
    verifyAndGetLogSegmentValidSize(blobStoreStats, new TimeRange(state.time.milliseconds(), 0L));
    verifyContainerStorageStatsAndGetTotalValidSize(blobStoreStats, state.time.milliseconds());
    assertEquals("Throttle count mismatch from expected value", throttleCountBeforeRequests,
        mockThrottler.throttleCount.get());
    blobStoreStats.close();
  }
Test to verify new entries after a scan are properly counted to keep the current {@link ScanResults} relevant
before the next scan. In addition, verify that requests within the coverage are served without triggering any scan.
@throws StoreException
@throws InterruptedException
ambry-store/src/test/java/com/github/ambry/store/BlobStoreStatsTest.java