@Test
  public void cloudReplicaRemovalTest() 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();
    PartitionId localPartition = storageManager.getLocalPartitions().iterator().next();
    // 1. add cloud replica first for subsequent removal test
    mockHelixParticipant.onPartitionBecomeLeaderFromStandby(localPartition.toPathString());
    String replicaPath = Cloud_Replica_Keyword + File.separator + localPartition.toPathString() + File.separator
        + localPartition.toPathString();
    RemoteReplicaInfo remoteReplicaInfo =
        recoveryManager.getRemoteReplicaInfo(localPartition, vcrNode.getHostname(), replicaPath);
    assertNotNull("Remote replica info should not be null", remoteReplicaInfo);
    assertEquals("There should be only one cloud replica thread created", 1,
        TestUtils.getAllThreadsByThisName(REPLICA_THREAD_PREFIX).size());

    // 2. before removing cloud replica of local partition let's remove a non-existent partition first
    mockHelixParticipant.onPartitionBecomeStandbyFromLeader(NEW_PARTITION_NAME);
    // ensure there is no change in replica thread
    assertEquals("There should be only one cloud replica thread created", 1,
        TestUtils.getAllThreadsByThisName(REPLICA_THREAD_PREFIX).size());

    // 3. remove the cloud replica by calling Leader-To-Standby transition on local partition
    mockHelixParticipant.onPartitionBecomeStandbyFromLeader(localPartition.toPathString());
    // ensure that the remote replica info has been successfully removed from replica thread
    assertNull("Cloud replica should be removed and no thread is assigned to it", remoteReplicaInfo.getReplicaThread());
    recoveryManager.shutdown();
    storageManager.shutdown();
  }
Test both success and failure cases when removing cloud replica.
@throws Exception
ambry-cloud/src/test/java/com/github/ambry/cloud/RecoveryManagerTest.java