@Test
  public void testResetPartitions() throws Exception {
    assumeTrue(stateModelDef.equals(ClusterMapConfig.AMBRY_STATE_MODEL_DEF));
    ClusterMapConfig clusterMapConfig = new ClusterMapConfig(new VerifiableProperties(props));
    MetricRegistry metricRegistry = new MetricRegistry();
    HelixParticipant helixParticipant =
        new HelixParticipant(mock(HelixClusterManager.class), clusterMapConfig, new HelixFactory(), metricRegistry,
            getDefaultZkConnectStr(clusterMapConfig), true);
    // Mock a state change listener to throw an exception
    PartitionStateChangeListener listener = mock(PartitionStateChangeListener.class);
    doThrow(new StateTransitionException("error", StateTransitionException.TransitionErrorCode.BootstrapFailure)).when(
        listener).onPartitionBecomeBootstrapFromOffline(anyString());
    helixParticipant.registerPartitionStateChangeListener(StateModelListenerType.StatsManagerListener, listener);

    // participate
    helixParticipant.participateAndBlockStateTransition(Collections.emptyList(), null, null);
    helixParticipant.unblockStateTransition();
    HelixManager manager = helixParticipant.getHelixManager();
    ClusterMap cm = new StaticClusterManager(testPartitionLayout.partitionLayout, dcName, new MetricRegistry());
    List<? extends ReplicaId> replicaIds =
        cm.getReplicaIds(cm.getDataNodeId("localhost", clusterMapConfig.clusterMapPort));
    String resource = "10000"; // There is only one resource
    // send state transition messages, this is controller logic, just put it down here to make sure the pipeline
    // of state transition works
    sendStateTransitionMessages(manager, resource, replicaIds, "OFFLINE", "BOOTSTRAP");
    // sleep some time so the state transition can happen
    Thread.sleep(1000);
    getNumberOfReplicaInStateFromMetric("offline", metricRegistry);
    // All replicas should at error state now
    assertEquals(replicaIds.size(), getNumberOfReplicaInStateFromMetric("error", metricRegistry));

    // reset one partition
    assertTrue(helixParticipant.resetPartitionState("" + replicaIds.get(0).getPartitionId().getId()));
    Thread.sleep(500);
    assertEquals(1, getNumberOfReplicaInStateFromMetric("offline", metricRegistry));
    assertEquals(replicaIds.size() - 1, getNumberOfReplicaInStateFromMetric("error", metricRegistry));

    // reset remaining partitions
    List<String> remainingPartitions = replicaIds.subList(1, replicaIds.size())
        .stream()
        .map(ReplicaId::getPartitionId)
        .map(PartitionId::getId)
        .map(String::valueOf)
        .collect(Collectors.toList());
    assertTrue(helixParticipant.resetPartitionState(remainingPartitions));
    Thread.sleep(500);
    assertEquals(replicaIds.size(), getNumberOfReplicaInStateFromMetric("offline", metricRegistry));
    assertEquals(0, getNumberOfReplicaInStateFromMetric("error", metricRegistry));

    helixParticipant.close();
  }
Test reset partition state methods
@throws Exception
ambry-clustermap/src/test/java/com/github/ambry/clustermap/HelixParticipantTest.java