@Test
  public void testCompactionExecutorHappyPath() throws Exception {
    int numStores = 5;
    List<BlobStore> stores = new ArrayList<>();
    // one store with nothing to compact isn't going to get compact calls.
    // since we are using mock time, wait for compact calls to arrive twice to ensure the time based scheduling works
    CountDownLatch compactCallsCountdown = new CountDownLatch(2 * (numStores - 1));
    properties.setProperty("store.compaction.triggers", ALL_COMPACTION_TRIGGERS);
    config = new StoreConfig(new VerifiableProperties(properties));
    MetricRegistry metricRegistry = new MetricRegistry();
    StoreMetrics metrics = new StoreMetrics(metricRegistry);
    MockBlobStore lastStore = null;
    for (int i = 0; i < numStores; i++) {
      MockBlobStore store = new MockBlobStore(config, metrics, time, compactCallsCountdown, null);
      // one store should not have any segments to compact
      store.details = i == 0 ? null : generateRandomCompactionDetails(i);
      stores.add(store);
      lastStore = store;
    }
    compactionManager =
        new CompactionManager(MOUNT_PATH, config, stores, new StorageManagerMetrics(metricRegistry), time);
    compactionManager.enable();
    assertNotNull("Compaction thread should be created",
        TestUtils.getThreadByThisName(CompactionManager.THREAD_NAME_PREFIX));
    assertTrue("Compaction calls did not come within the expected time",
        compactCallsCountdown.await(1, TimeUnit.SECONDS));
    assertTrue("Compaction Executor should be running", compactionManager.isCompactionExecutorRunning());
    for (int i = 0; i < numStores; i++) {
      MockBlobStore store = (MockBlobStore) stores.get(i);
      if (store.callOrderException != null) {
        throw store.callOrderException;
      }
      if (i > 0) {
        assertTrue("Compact was not called", store.compactCalled);
      } else {
        // should not call for i == 0 because there are no compaction details.
        assertFalse("Compact should not have been called", store.compactCalled);
      }
    }
    // test scheduleNextForCompaction()
    lastStore.compactCallsCountdown = new CountDownLatch(1);
    lastStore.compactCalled = false;
    assertTrue("Should schedule compaction", compactionManager.scheduleNextForCompaction(lastStore));
    assertTrue("Compaction call did not come within the expected time",
        lastStore.compactCallsCountdown.await(1, TimeUnit.HOURS));
    if (lastStore.callOrderException != null) {
      throw lastStore.callOrderException;
    }
    assertTrue("compact() should have been called", lastStore.compactCalled);
    compactionManager.disable();
    compactionManager.awaitTermination();
    assertFalse("Compaction thread should not be running", compactionManager.isCompactionExecutorRunning());
  }
Tests that compaction is triggered on all stores provided they do not misbehave. Also includes a store that is
not ready for compaction. Ensures that {@link BlobStore#maybeResumeCompaction(byte[])} is called before
{@link BlobStore#compact(CompactionDetails, byte[])} is called.
@throws Exception
ambry-store/src/test/java/com/github/ambry/store/CompactionManagerTest.java