@Test
  public void testGetAndSetReplicaStoppedState() throws Exception {
    //setup HelixParticipant, HelixParticipantDummy and dependencies
    ClusterMapConfig clusterMapConfig = new ClusterMapConfig(new VerifiableProperties(props));
    String instanceName = ClusterMapUtils.getInstanceName("localhost", clusterMapConfig.clusterMapPort);
    HelixParticipant helixParticipant =
        new HelixParticipant(mock(HelixClusterManager.class), clusterMapConfig, new HelixFactory(),
            new MetricRegistry(), getDefaultZkConnectStr(clusterMapConfig), true);
    ZKHelixAdmin helixAdmin = new ZKHelixAdmin("localhost:" + zkInfo.getPort());
    DataNodeConfig dataNodeConfig = getDataNodeConfigInHelix(helixAdmin, instanceName);

    //Make sure the current stoppedReplicas list is empty
    List<String> stoppedReplicas = helixParticipant.getStoppedReplicas();
    assertEquals("stoppedReplicas list should be empty", Collections.emptyList(), stoppedReplicas);

    String listName = "stoppedReplicas list";
    Set<String> localPartitionNames = new HashSet<>();
    dataNodeConfig.getDiskConfigs()
        .values()
        .forEach(diskConfig -> localPartitionNames.addAll(diskConfig.getReplicaConfigs().keySet()));
    String[] partitionIds = new String[3];
    for (int i = 0; i < partitionIds.length; ++i) {
      partitionIds[i] = localPartitionNames.iterator().next();
      localPartitionNames.remove(partitionIds[i]);
    }
    ReplicaId replicaId1 = createMockAmbryReplica(partitionIds[0]);
    ReplicaId replicaId2 = createMockAmbryReplica(partitionIds[1]);
    ReplicaId replicaId3 = createMockAmbryReplica(partitionIds[2]);
    //Check that invoking setReplicaStoppedState with a non-AmbryReplica ReplicaId throws an IllegalArgumentException
    ReplicaId nonAmbryReplica = createMockNotAmbryReplica(partitionIds[1]);
    try {
      helixParticipant.setReplicaStoppedState(Collections.singletonList(nonAmbryReplica), true);
      fail("Expected an IllegalArgumentException here");
    } catch (IllegalArgumentException e) {
      // expected. Nothing to do.
    }

    //Check that invoking setReplicaStoppedState adds the replicaId1, replicaId2 to the list of stopped replicas
    helixParticipant.setReplicaStoppedState(Arrays.asList(replicaId1, replicaId2), true);
    stoppedReplicas = helixParticipant.getStoppedReplicas();
    listIsExpectedSize(stoppedReplicas, 2, listName);
    assertTrue(stoppedReplicas.contains(replicaId1.getPartitionId().toPathString()));
    assertTrue(stoppedReplicas.contains(replicaId2.getPartitionId().toPathString()));

    //Invoke setReplicaStoppedState to add replicaId1, replicaId2 again, should be no-op
    helixParticipant.setReplicaStoppedState(Arrays.asList(replicaId1, replicaId2), true);
    listIsExpectedSize(helixParticipant.getStoppedReplicas(), 2, listName);

    //Add replicaId1 again as well as replicaId3 to ensure new replicaId is correctly added and no duplicates in the stopped list
    helixParticipant.setReplicaStoppedState(Arrays.asList(replicaId1, replicaId3), true);
    stoppedReplicas = helixParticipant.getStoppedReplicas();
    listIsExpectedSize(stoppedReplicas, 3, listName);
    assertTrue(stoppedReplicas.contains(replicaId1.getPartitionId().toPathString()));
    assertTrue(stoppedReplicas.contains(replicaId2.getPartitionId().toPathString()));
    assertTrue(stoppedReplicas.contains(replicaId3.getPartitionId().toPathString()));

    //Check that invoking setReplicaStoppedState with markStop == false removes replicaId1, replicaId2 from stopped list
    helixParticipant.setReplicaStoppedState(Arrays.asList(replicaId1, replicaId2), false);
    stoppedReplicas = helixParticipant.getStoppedReplicas();
    listIsExpectedSize(stoppedReplicas, 1, listName);
    assertTrue(stoppedReplicas.contains(replicaId3.getPartitionId().toPathString()));
    assertFalse(stoppedReplicas.contains(replicaId2.getPartitionId().toPathString()));
    assertFalse(stoppedReplicas.contains(replicaId1.getPartitionId().toPathString()));

    //Removing replicaIds which have already been removed doesn't hurt anything and will not update InstanceConfig in Helix
    helixParticipant.setReplicaStoppedState(Arrays.asList(replicaId1, replicaId2), false);
    stoppedReplicas = helixParticipant.getStoppedReplicas();
    listIsExpectedSize(stoppedReplicas, 1, listName);
    assertTrue(stoppedReplicas.contains(replicaId3.getPartitionId().toPathString()));

    //Removing all replicas (including replica not in the list) yields expected behavior
    helixParticipant.setReplicaStoppedState(Arrays.asList(replicaId2, replicaId3), false);
    stoppedReplicas = helixParticipant.getStoppedReplicas();
    listIsExpectedSize(stoppedReplicas, 0, listName);
    helixAdmin.close();
    helixParticipant.close();
  }
Tests setReplicaStoppedState method for {@link HelixParticipant}
@throws Exception
ambry-clustermap/src/test/java/com/github/ambry/clustermap/HelixParticipantTest.java