@Test
  public void testCompactionWithMisbehavingStores() throws Exception {
    int numStores = 5;
    List<BlobStore> stores = new ArrayList<>();
    // one store that isn't started isn't going to get compact calls.
    // another store that throws an exception on resumeCompaction() isn't going to get a compact call.
    // the new added store in storesDisabledCompaction isn't going to get compact calls.
    CountDownLatch compactCallsCountdown = new CountDownLatch(numStores - 3);
    properties.setProperty("store.compaction.triggers", ALL_COMPACTION_TRIGGERS);
    properties.setProperty("store.compaction.check.frequency.in.hours", Integer.toString(100));
    config = new StoreConfig(new VerifiableProperties(properties));
    MetricRegistry metricRegistry = new MetricRegistry();
    StoreMetrics metrics = new StoreMetrics(metricRegistry);
    for (int i = 0; i < numStores; i++) {
      MockBlobStore store = new MockBlobStore(config, metrics, time, compactCallsCountdown, null);
      store.details = generateRandomCompactionDetails(2);
      store.storeId = String.valueOf(i);
      if (i == 0) {
        // one store should not be started
        store.started = false;
      } else if (i == 1) {
        // one store should throw on resumeCompaction()
        store.exceptionToThrowOnResume = new RuntimeException("Misbehaving store");
      } else if (i == 3) {
        // one store should throw on compact()
        store.exceptionToThrowOnCompact = new RuntimeException("Misbehaving store");
      }
      stores.add(store);
    }
    // This CountDownLatch is to ensure that current test thread won't advance until the exception thrown on compact()
    // is caught and processed. (Mainly to address race condition that store 3 is the last store being checked for
    // compaction. The compact() method is called but exception hasn't been captured. At this point of time, cpu switches
    // back to this test thread and checks scheduleNextForCompaction() on each store. Since exception of store 3 hasn't
    // been captured, store 3 is not in storesToSkip set and assert fails)
    MockStorageManagerMetrics mockMetrics = new MockStorageManagerMetrics(metricRegistry);
    // The reason to set latch count = numStore + 2 is, stores[1 - 4] plus new added will be checked if resume compaction
    // (store 0 won't be checked because it is not started). So markCompactionStop() is called on 4 + 1 stores when checking
    // whether to resume compaction. After that only store 3 and store 4 are eligible to perform compact() and call
    // markCompactionStop() twice.
    CountDownLatch compactStopCountdown = new CountDownLatch(4 + 1 + 2);
    mockMetrics.compactStopCountdown = compactStopCountdown;
    // using real time here so that compaction is not scheduled more than once for a store during the test unless
    // asked for.
    compactionManager = new CompactionManager(MOUNT_PATH, config, stores, mockMetrics, SystemTime.getInstance());
    // disable the third blobstore for compaction before starting the CompactionExecutor thread
    MockBlobStore controlCompactionTestStore = (MockBlobStore) stores.get(2);
    compactionManager.controlCompactionForBlobStore(controlCompactionTestStore, false);
    // add a new blobstore into compaction manager but not started yet.
    MockBlobStore newAddedStore =
        new MockBlobStore(config, metrics, time, compactCallsCountdown, generateRandomCompactionDetails(2));
    newAddedStore.storeId = "newStore";
    compactionManager.addBlobStore(newAddedStore);
    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 stop latch didn't count down to 0 within expected time",
        compactStopCountdown.await(1, TimeUnit.SECONDS));
    for (int i = 0; i < numStores; i++) {
      MockBlobStore store = (MockBlobStore) stores.get(i);
      if (store.callOrderException != null) {
        throw store.callOrderException;
      }
      if (i > 2) {
        assertTrue("Compact was not called", store.compactCalled);
      } else {
        // should not call for i == 0 because store has not been started.
        // should not call for i == 1 because resumeCompaction() would have marked this as a misbehaving store.
        // should not call for i == 2 because store has been disabled for compaction.
        assertFalse("Compact should not have been called", store.compactCalled);
      }
    }
    // ensure compact is not called on new added store
    assertFalse("Compact should not have been called", newAddedStore.compactCalled);

    // set all compact called to false
    for (BlobStore store : stores) {
      ((MockBlobStore) store).compactCalled = false;
    }
    // stores that are not started, disabled or failed on resumeCompaction() and compact() cannot be scheduled for compaction
    for (int i = 0; i < numStores; i++) {
      MockBlobStore store = (MockBlobStore) stores.get(i);
      if (i < 4) {
        assertFalse("Should not schedule compaction for " + store, compactionManager.scheduleNextForCompaction(store));
        assertFalse("compact() should not have been called on " + store, store.compactCalled);
      } else {
        store.compactCallsCountdown = new CountDownLatch(1);
        assertFalse("compactCalled should be reset", store.compactCalled);
        assertTrue("Should schedule compaction for " + store, compactionManager.scheduleNextForCompaction(store));
        assertTrue("Compaction call did not come within the expected time",
            store.compactCallsCountdown.await(1, TimeUnit.SECONDS));
        if (store.callOrderException != null) {
          throw store.callOrderException;
        }
        assertTrue("compact() should have been called on " + store, store.compactCalled);
      }
    }

    // disable Compaction Manager for restart
    compactionManager.disable();
    compactionManager.awaitTermination();
    assertFalse("Compaction thread should not be running", compactionManager.isCompactionExecutorRunning());
    compactCallsCountdown = new CountDownLatch(3);
    // set all compact called to false
    for (BlobStore store : stores) {
      ((MockBlobStore) store).compactCalled = false;
      ((MockBlobStore) store).resumeCompactionCalled = false;
      ((MockBlobStore) store).compactCallsCountdown = compactCallsCountdown;
    }
    newAddedStore.compactCallsCountdown = compactCallsCountdown;
    compactionManager.controlCompactionForBlobStore(controlCompactionTestStore, true);
    compactionManager.controlCompactionForBlobStore(newAddedStore, true);
    // restart Compaction Manager to trigger Periodic Compaction
    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));
    for (int i = 0; i < numStores; i++) {
      MockBlobStore store = (MockBlobStore) stores.get(i);
      if (store.callOrderException != null) {
        throw store.callOrderException;
      }
      if (i == 2 || i == 4) {
        assertTrue("Compact was not called", store.compactCalled);
      } else {
        // should not call for i == 0 because store has not been started.
        // should not call for i == 1 because resumeCompaction() would have marked this as a misbehaving store.
        // should not call for i == 3 because it encountered exception during last compaction.
        assertFalse("Compact should not have been called", store.compactCalled);
      }
    }
    // ensure compact is called for new added store
    assertTrue("Compact was not called", newAddedStore.compactCalled);
    compactionManager.disable();
    compactionManager.awaitTermination();
    assertFalse("Compaction thread should not be running", compactionManager.isCompactionExecutorRunning());
  }
Tests that compaction proceeds on all non misbehaving stores even in the presence of some misbehaving stores
(not started/throwing exceptions).
@throws Exception
ambry-store/src/test/java/com/github/ambry/store/CompactionManagerTest.java