private void testHardDeletePauseResume(boolean reloadIndex) throws InterruptedException, IOException, StoreException {
    state.properties.put("store.deleted.message.retention.minutes", Integer.toString(60));
    state.properties.put("store.hard.delete.operations.bytes.per.sec", Integer.toString(Integer.MAX_VALUE / 10));
    state.properties.setProperty("store.data.flush.interval.seconds", "3600");
    // enable daemon thread to run hard deletes
    state.properties.put("store.enable.hard.delete", "true");
    state.reloadIndex(true, false);
    assertTrue("Hard delete is not enabled", state.index.hardDeleter.isRunning());
    // IndexSegment still uses real time so advance time so that it goes 2 days past the real time.
    state.advanceTime(TimeUnit.DAYS.toMillis(2));
    long expectedProgress = state.index.getAbsolutePositionInLogForOffset(state.logOrder.lastKey());
    // give it some time so that hard delete completes one cycle
    waitUntilExpectedProgress(expectedProgress, 5000);
    state.verifyEntriesForHardDeletes(state.deletedKeys);
    Set<MockId> idsDeleted = new HashSet<>();
    // delete two entries
    state.addPutEntries(2, CuratedLogIndexState.PUT_RECORD_SIZE, Utils.Infinite_Time);
    Set<MockId> idsToDelete = new HashSet<>();
    idsToDelete.add(state.getIdToDeleteFromIndexSegment(state.referenceIndex.lastKey(), false));
    state.addPutEntries(2, CuratedLogIndexState.PUT_RECORD_SIZE, Utils.Infinite_Time);
    idsToDelete.add(state.getIdToDeleteFromIndexSegment(state.referenceIndex.lastKey(), false));
    for (MockId id : idsToDelete) {
      state.addDeleteEntry(id);
    }
    idsDeleted.addAll(idsToDelete);
    // pause hard delete
    state.index.hardDeleter.pause();
    assertTrue("Hard deletes should have been paused ", state.index.hardDeleter.isPaused());
    assertTrue("Hard delete thread did not go into waiting state",
        TestUtils.waitUntilExpectedState(state.index.hardDeleteThread, Thread.State.WAITING,
            HardDeleter.HARD_DELETE_SLEEP_TIME_ON_CAUGHT_UP_MS + 1));

    // delete two entries
    state.addPutEntries(2, CuratedLogIndexState.PUT_RECORD_SIZE, Utils.Infinite_Time);
    idsToDelete.clear();
    idsToDelete.add(state.getIdToDeleteFromIndexSegment(state.referenceIndex.lastKey(), false));
    state.addPutEntries(2, CuratedLogIndexState.PUT_RECORD_SIZE, Utils.Infinite_Time);
    idsToDelete.add(state.getIdToDeleteFromIndexSegment(state.referenceIndex.lastKey(), false));
    for (MockId id : idsToDelete) {
      state.addDeleteEntry(id);
    }
    idsDeleted.addAll(idsToDelete);

    if (reloadIndex) {
      state.reloadIndex(true, true);
      assertTrue("Hard delete thread did not go into waiting state",
          TestUtils.waitUntilExpectedState(state.index.hardDeleteThread, Thread.State.WAITING,
              HardDeleter.HARD_DELETE_SLEEP_TIME_ON_CAUGHT_UP_MS + 1));
      idsToDelete.clear();
      state.addPutEntries(2, CuratedLogIndexState.PUT_RECORD_SIZE, Utils.Infinite_Time);
      idsToDelete.add(state.getIdToDeleteFromIndexSegment(state.referenceIndex.lastKey(), false));
      state.addPutEntries(2, CuratedLogIndexState.PUT_RECORD_SIZE, Utils.Infinite_Time);
      idsToDelete.add(state.getIdToDeleteFromIndexSegment(state.referenceIndex.lastKey(), false));
      for (MockId id : idsToDelete) {
        state.addDeleteEntry(id);
      }
      idsDeleted.addAll(idsToDelete);
    }

    // advance time so that deleted entries becomes eligible to be hard deleted
    state.advanceTime(TimeUnit.DAYS.toMillis(3));
    // resume and verify new entries have been hard deleted
    state.index.hardDeleter.resume();
    assertFalse("Hard deletes should have been resumed ", state.index.hardDeleter.isPaused());
    expectedProgress = state.index.getAbsolutePositionInLogForOffset(state.logOrder.lastKey());
    // after resuming. hard deletes should progress. Give it some time to hard delete next range
    waitUntilExpectedProgress(expectedProgress, 5000);
    state.verifyEntriesForHardDeletes(idsDeleted);
  }
Utility to test hard delete pause and resume with or without reload between pause and resume
@param reloadIndex {@code true} if index has to be reloaded inbetween pause and resume. {@code false} otherwise
@throws InterruptedException
@throws IOException
@throws StoreException
ambry-store/src/test/java/com/github/ambry/store/IndexTest.java