@Test
  public void cloudReplicaAdditionTest() throws Exception {
    StorageManager storageManager =
        new StorageManager(storeConfig, new DiskManagerConfig(verifiableProperties), Utils.newScheduler(1, true),
            clusterMap.getMetricRegistry(), null, clusterMap, currentNode, null,
            Collections.singletonList(mockHelixParticipant), new MockTime(), null,
            new InMemAccountService(false, false));
    RecoveryManager recoveryManager =
        new RecoveryManager(replicationConfig, clusterMapConfig, storeConfig, storageManager,
            storeKeyFactory, clusterMap, mockScheduler, currentNode, mock(NetworkClientFactory.class),
            clusterMap.getMetricRegistry(), null, storeKeyConverterFactory, serverConfig.serverMessageTransformer,
            mockClusterSpectator, mockHelixParticipant);
    storageManager.start();
    recoveryManager.start();
    mockClusterSpectator.spectate();
    // 1. test adding cloud replica that is not present locally
    mockHelixParticipant.onPartitionBecomeLeaderFromStandby(NEW_PARTITION_NAME);
    assertNull("Cloud replica thread should not be created", TestUtils.getThreadByThisName(REPLICA_THREAD_PREFIX));
    // create a new partition and add corresponding store in storage manager
    PartitionId newPartition =
        new MockPartitionId(Long.parseLong(NEW_PARTITION_NAME), MockClusterMap.DEFAULT_PARTITION_CLASS,
            clusterMap.getDataNodes(), 0);
    ReplicaId replicaToAdd = newPartition.getReplicaIds().get(0);
    assertTrue("Adding new store should succeed", storageManager.addBlobStore(replicaToAdd));
    // 2. we deliberately shut down the store to induce failure when adding cloud replica
    storageManager.shutdownBlobStore(newPartition);
    mockHelixParticipant.onPartitionBecomeLeaderFromStandby(NEW_PARTITION_NAME);
    assertNull("Cloud replica thread should not be created", TestUtils.getThreadByThisName(REPLICA_THREAD_PREFIX));
    storageManager.startBlobStore(newPartition);
    // 3. mock success case
    mockHelixParticipant.onPartitionBecomeLeaderFromStandby(NEW_PARTITION_NAME);
    assertNotNull("Cloud replica thread should be created for DC1",
        TestUtils.getThreadByThisName(REPLICA_THREAD_PREFIX));
    recoveryManager.shutdown();
    storageManager.shutdown();
  }
Test both success and failure cases when adding cloud replica
@throws Exception
ambry-cloud/src/test/java/com/github/ambry/cloud/RecoveryManagerTest.java