@Test
  public void skipStoppedStoresTest() throws Exception {
    MockDataNodeId dataNode = clusterMap.getDataNodes().get(0);
    List<ReplicaId> replicas = clusterMap.getReplicaIds(dataNode);
    ClusterParticipant mockParticipant = new MockClusterParticipant();
    mockParticipant.setReplicaStoppedState(Collections.singletonList(replicas.get(0)), true);
    StorageManager storageManager =
        createStorageManager(dataNode, metricRegistry, Collections.singletonList(mockParticipant));
    storageManager.start();
    assertEquals("There should be no unexpected partitions reported", 0, getNumUnrecognizedPartitionsReported());
    for (int i = 0; i < replicas.size(); ++i) {
      PartitionId id = replicas.get(i).getPartitionId();
      if (i == 0) {
        assertNull("Store should be null because stopped stores will be skipped and will not be started",
            storageManager.getStore(id, false));
        assertFalse("Compaction should not be scheduled", storageManager.scheduleNextForCompaction(id));
      } else {
        Store store = storageManager.getStore(id, false);
        assertTrue("Store should be started", store.isStarted());
        assertTrue("Compaction should be scheduled", storageManager.scheduleNextForCompaction(id));
      }
    }
    shutdownAndAssertStoresInaccessible(storageManager, replicas);
  }
Test the stopped stores are correctly skipped and not started during StorageManager's startup.
ambry-store/src/test/java/com/github/ambry/store/StorageManagerTest.java