@Test
  public void testCompactionAllSlowWorkers() throws InterruptedException {
    int numPartitions = 5000;
    int numSlowWorkers = cloudConfig.cloudCompactionNumThreads; // all slow workers
    addPartitionsToCompact(numPartitions);
    // mockDest is used inside compactor but Mockito cannot infer this.
    // Use lenient() to avoid UnnecessaryStubbingException.
    // Emulate slow worker for some partitions
    CountDownLatch fastWorkerLatch = new CountDownLatch(0);
    CountDownLatch slowWorkerLatch = new CountDownLatch(numSlowWorkers);
    try {
      Mockito.lenient().when(mockDest.compactPartition(any(), any()))
          .thenAnswer((Answer<Integer>) invocation ->
              slowWorker(invocation.getArgument(0), numSlowWorkers, fastWorkerLatch, slowWorkerLatch));
    } catch (CloudStorageException e) {
      throw new RuntimeException(e);
    }
    cloudCompactionScheduler.scheduleWithFixedDelay(compactor, cloudConfig.cloudBlobCompactionStartupDelaySecs,
        cloudConfig.cloudBlobCompactionIntervalHours, TimeUnit.HOURS);
    slowWorkerLatch.await();
    Mockito.lenient().when(mockDest.stopCompaction()).thenAnswer(invocation -> {
      logger.info("[TEST] Number of active tasks = {}", compactor.getNumActiveCompactionTasks());
      logger.info("[TEST] Number of waiting tasks = {}", compactor.getNumWaitingCompactionTasks());
      logger.info("[TEST] Number of completed tasks = {}", compactor.getNumCompletedCompactionTasks());
      logger.info("[TEST] Number of all tasks = {}", compactor.getNumAllCompactionTasks());
      assertEquals(numSlowWorkers, compactor.getNumActiveCompactionTasks());
      assertEquals(0, compactor.getNumWaitingCompactionTasks()); // The queue has been drained by now
      return true;
    });
    shutdownCompactionWorkers(compactor);
  }
Tests compaction shutdown for slow workers
@throws InterruptedException
ambry-cloud/src/test/java/com/github/ambry/cloud/CloudStorageCompactorTest.java