@Test
  public void scheduleAndControlCompactionTest() throws Exception {
    MockDataNodeId dataNode = clusterMap.getDataNodes().get(0);
    List<ReplicaId> replicas = clusterMap.getReplicaIds(dataNode);
    List<MockDataNodeId> dataNodes = new ArrayList<>();
    dataNodes.add(dataNode);
    MockPartitionId invalidPartition =
        new MockPartitionId(Long.MAX_VALUE, MockClusterMap.DEFAULT_PARTITION_CLASS, dataNodes, 0);
    List<? extends ReplicaId> invalidPartitionReplicas = invalidPartition.getReplicaIds();
    StorageManager storageManager = createStorageManager(dataNode, metricRegistry, null);
    storageManager.start();
    assertEquals("There should be 1 unexpected partition reported", 1, getNumUnrecognizedPartitionsReported());
    // add invalid replica id
    replicas.add(invalidPartitionReplicas.get(0));
    for (int i = 0; i < replicas.size(); i++) {
      ReplicaId replica = replicas.get(i);
      PartitionId id = replica.getPartitionId();
      if (i == replicas.size() - 1) {
        assertFalse("Schedule compaction should fail", storageManager.scheduleNextForCompaction(id));
        assertFalse("Disable compaction should fail", storageManager.controlCompactionForBlobStore(id, false));
        assertFalse("Enable compaction should fail", storageManager.controlCompactionForBlobStore(id, true));
      } else {
        assertTrue("Enable compaction should succeed", storageManager.controlCompactionForBlobStore(id, true));
        assertTrue("Schedule compaction should succeed", storageManager.scheduleNextForCompaction(id));
      }
    }
    ReplicaId replica = replicas.get(0);
    PartitionId id = replica.getPartitionId();
    assertTrue("Disable compaction should succeed", storageManager.controlCompactionForBlobStore(id, false));
    assertFalse("Schedule compaction should fail", storageManager.scheduleNextForCompaction(id));
    assertTrue("Enable compaction should succeed", storageManager.controlCompactionForBlobStore(id, true));
    assertTrue("Schedule compaction should succeed", storageManager.scheduleNextForCompaction(id));

    replica = replicas.get(1);
    id = replica.getPartitionId();
    assertTrue("Schedule compaction should succeed", storageManager.scheduleNextForCompaction(id));
    replicas.remove(replicas.size() - 1);
    shutdownAndAssertStoresInaccessible(storageManager, replicas);
  }
Tests that schedule compaction and control compaction in StorageManager
@throws Exception
ambry-store/src/test/java/com/github/ambry/store/StorageManagerTest.java