@Test
  public void testAmbryPartitionStateModelWithDuplicatePartitionIds() {
    assumeTrue(stateModelDef.equals(ClusterMapConfig.AMBRY_STATE_MODEL_DEF) && !ignoreDownwardStateTransition);
    MetricRegistry metricRegistry = new MetricRegistry();
    MockHelixParticipant.metricRegistry = metricRegistry;
    DataNodeConfig mockDataNodeConfig = Mockito.mock(DataNodeConfig.class);
    Set<String> disabledPartitionSet = new HashSet<>();
    Set<String> enabledPartitionSet = new HashSet<>();
    when(mockDataNodeConfig.getDisabledReplicas()).thenReturn(disabledPartitionSet);
    DataNodeConfigSource mockConfigSource = Mockito.mock(DataNodeConfigSource.class);
    when(mockConfigSource.get(anyString())).thenReturn(mockDataNodeConfig);
    HelixAdmin mockHelixAdmin = Mockito.mock(HelixAdmin.class);
    InstanceConfig mockInstanceConfig = Mockito.mock(InstanceConfig.class);
    doAnswer(invocation -> {
      String partitionName = invocation.getArgument(1);
      boolean enable = invocation.getArgument(2);
      if (enable) {
        enabledPartitionSet.add(partitionName);
      }
      return null;
    }).when(mockInstanceConfig).setInstanceEnabledForPartition(any(), any(), anyBoolean());
    when(mockHelixAdmin.getInstanceConfig(anyString(), anyString())).thenReturn(mockInstanceConfig);
    when(mockHelixAdmin.setInstanceConfig(anyString(), anyString(), any())).thenReturn(true);
    HelixManager mockHelixManager = Mockito.mock(HelixManager.class);
    when(mockHelixManager.getClusterManagmentTool()).thenReturn(mockHelixAdmin);
    MockHelixManagerFactory.overrideGetHelixManager = true;
    MockHelixParticipant.mockHelixFactory = new MockHelixManagerFactory(mockConfigSource, mockHelixManager);
    MockHelixParticipant mockHelixParticipant = new MockHelixParticipant(config);
    HelixParticipantMetrics participantMetrics = mockHelixParticipant.getHelixParticipantMetrics();

    String resourceName = "0";
    String partitionName = "1";
    String newResourceName = "10";
    Message mockMessage = Mockito.mock(Message.class);
    when(mockMessage.getPartitionName()).thenReturn(partitionName);
    when(mockMessage.getResourceName()).thenReturn(resourceName);

    Message newMockMessage = Mockito.mock(Message.class);
    when(newMockMessage.getPartitionName()).thenReturn(partitionName);
    when(newMockMessage.getResourceName()).thenReturn(newResourceName);

    HelixClusterManager clusterManager = mock(HelixClusterManager.class);

    // Test case 1. clustermap returns the newResourceName when retrieving resource name for partition
    when(clusterManager.getResourceForPartitionInLocalDc(anyString())).thenReturn(
        Collections.singleton(newResourceName));
    AmbryPartitionStateModel stateModel =
        new AmbryPartitionStateModel(resourceName, partitionName, mockHelixParticipant, config, clusterManager);
    AmbryPartitionStateModel newStateModel =
        new AmbryPartitionStateModel(newResourceName, partitionName, mockHelixParticipant, config, clusterManager);

    // resource move to bootstrap then new resource start transition
    mockHelixParticipant.setInitialLocalPartitions(new HashSet<>(Collections.singletonList(partitionName)));
    assertStateCount(Arrays.asList("offline"), Arrays.asList(1), metricRegistry);
    // resource: OFFLINE -> BOOTSTRAP, shouldn't work, partition belongs to new resource name
    stateModel.onBecomeBootstrapFromOffline(mockMessage, null);
    assertStateCount(Arrays.asList("offline", "bootstrap"), Arrays.asList(1, 0), metricRegistry);
    // resource: BOOTSTRAP -> STANDBY, shouldn't work, partition belongs to new resource name
    stateModel.onBecomeStandbyFromBootstrap(mockMessage, null);
    assertStateCount(Arrays.asList("offline", "bootstrap", "standby"), Arrays.asList(1, 0, 0), metricRegistry);
    // new resource: OFFLINE -> BOOTSTRAP, should work
    newStateModel.onBecomeBootstrapFromOffline(newMockMessage, null);
    assertStateCount(Arrays.asList("offline", "bootstrap", "standby"), Arrays.asList(0, 1, 0), metricRegistry);
    // resource: STANDBY -> LEADER, should not work
    stateModel.onBecomeLeaderFromStandby(mockMessage, null);
    assertStateCount(Arrays.asList("offline", "bootstrap", "standby", "leader"), Arrays.asList(0, 1, 0, 0),
        metricRegistry);
    // resource: LEADER -> STANDBY, should not work
    stateModel.onBecomeStandbyFromLeader(mockMessage, null);
    assertStateCount(Arrays.asList("offline", "bootstrap", "standby", "leader"), Arrays.asList(0, 1, 0, 0),
        metricRegistry);
    // resource: STANDBY -> INACTIVE, should not work
    stateModel.onBecomeInactiveFromStandby(mockMessage, null);
    assertStateCount(Arrays.asList("offline", "bootstrap", "standby", "inactive"), Arrays.asList(0, 1, 0, 0),
        metricRegistry);
    // resource: INACTIVE -> OFFLINE, should not work
    stateModel.onBecomeOfflineFromInactive(mockMessage, null);
    assertStateCount(Arrays.asList("offline", "bootstrap", "inactive"), Arrays.asList(0, 1, 0), metricRegistry);
    // resource: OFFLINE -> DROPPED, should not work
    disabledPartitionSet.add(partitionName);
    stateModel.onBecomeDroppedFromOffline(mockMessage, null);
    assertStateCount(Arrays.asList("offline", "bootstrap"), Arrays.asList(0, 1), metricRegistry);
    assertEquals("Dropped count should not be updated", 0, participantMetrics.partitionDroppedCount.getCount());
    assertFalse("Partition should not be removed from disabled partition set", disabledPartitionSet.isEmpty());

    // resource: ERROR -> DROPPED, should not work
    stateModel.onBecomeDroppedFromError(mockMessage, null);
    assertEquals("Dropped count should not be updated", 0, participantMetrics.partitionDroppedCount.getCount());
    // resource: ERROR -> OFFLINE (this occurs when we use Helix API to reset certain partition in ERROR state), should not work
    stateModel.onBecomeOfflineFromError(mockMessage, null);
    assertStateCount(Arrays.asList("offline", "bootstrap"), Arrays.asList(0, 1), metricRegistry);

    // Now new resource would do the state transition
    // new resource: BOOTSTRAP -> STANDBY, should work
    newStateModel.onBecomeStandbyFromBootstrap(newMockMessage, null);
    assertStateCount(Arrays.asList("offline", "bootstrap", "standby"), Arrays.asList(0, 0, 1), metricRegistry);
    // new resource: STANDBY -> LEADER, should work
    newStateModel.onBecomeLeaderFromStandby(newMockMessage, null);
    assertStateCount(Arrays.asList("offline", "bootstrap", "standby", "leader"), Arrays.asList(0, 0, 0, 1),
        metricRegistry);
    // new resource: LEADER -> STANDBY, should work
    newStateModel.onBecomeStandbyFromLeader(newMockMessage, null);
    assertStateCount(Arrays.asList("offline", "bootstrap", "standby", "leader"), Arrays.asList(0, 0, 1, 0),
        metricRegistry);
    // new resource: STANDBY -> INACTIVE, should work
    newStateModel.onBecomeInactiveFromStandby(newMockMessage, null);
    assertStateCount(Arrays.asList("offline", "standby", "inactive"), Arrays.asList(0, 0, 1), metricRegistry);
    // new resource: INACTIVE -> OFFLINE, should work
    newStateModel.onBecomeOfflineFromInactive(newMockMessage, null);
    assertStateCount(Arrays.asList("offline", "inactive"), Arrays.asList(1, 0), metricRegistry);
    // new resource: OFFLINE -> DROPPED, should work
    disabledPartitionSet.add(partitionName);
    newStateModel.onBecomeDroppedFromOffline(newMockMessage, null);
    assertStateCount(Arrays.asList("offline"), Arrays.asList(0), metricRegistry);
    assertEquals("Dropped count should be updated", 1, participantMetrics.partitionDroppedCount.getCount());
    assertTrue("Partition should be removed from disabled partition set", disabledPartitionSet.isEmpty());
    assertEquals("Mismatch in enabled partition", partitionName, enabledPartitionSet.iterator().next());
    // new resource: ERROR -> DROPPED, should work
    newStateModel.onBecomeDroppedFromError(newMockMessage, null);
    assertEquals("Dropped count should be updated", 2, participantMetrics.partitionDroppedCount.getCount());
    // new resource: ERROR -> OFFLINE (this occurs when we use Helix API to reset certain partition in ERROR state)
    newStateModel.onBecomeOfflineFromError(newMockMessage, null);
    assertStateCount(Arrays.asList("offline"), Arrays.asList(1), metricRegistry);
    // reset method
    newStateModel.reset();
    assertStateCount(Arrays.asList("offline"), Arrays.asList(1), metricRegistry);
    // call reset method again to mock the case where same partition is reset multiple times during zk disconnection or shutdown
    newStateModel.reset();
    assertStateCount(Arrays.asList("offline"), Arrays.asList(1), metricRegistry);

    // Test case 2. clustermap returns the old resourceName when retrieving resource name for partition.
    when(clusterManager.getResourceForPartitionInLocalDc(anyString())).thenReturn(Collections.singleton(resourceName));
    stateModel =
        new AmbryPartitionStateModel(resourceName, partitionName, mockHelixParticipant, config, clusterManager);
    newStateModel =
        new AmbryPartitionStateModel(newResourceName, partitionName, mockHelixParticipant, config, clusterManager);

    // resource move to bootstrap then new resource start transition
    mockHelixParticipant.setInitialLocalPartitions(new HashSet<>(Collections.singletonList(partitionName)));
    // resource: OFFLINE -> BOOTSTRAP, should work
    stateModel.onBecomeBootstrapFromOffline(mockMessage, null);
    assertStateCount(Arrays.asList("offline", "bootstrap"), Arrays.asList(0, 1), metricRegistry);
    // resource: BOOTSTRAP -> STANDBY, should work
    stateModel.onBecomeStandbyFromBootstrap(mockMessage, null);
    assertStateCount(Arrays.asList("offline", "bootstrap", "standby"), Arrays.asList(0, 0, 1), metricRegistry);
    // resource: STANDBY -> LEADER, should work
    stateModel.onBecomeLeaderFromStandby(mockMessage, null);
    assertStateCount(Arrays.asList("offline", "bootstrap", "standby", "leader"), Arrays.asList(0, 0, 0, 1),
        metricRegistry);

    // new resource: OFFLINE -> BOOTSTRAP, should work
    newStateModel.onBecomeBootstrapFromOffline(newMockMessage, null);
    assertStateCount(Arrays.asList("offline", "bootstrap", "standby", "leader"), Arrays.asList(0, 1, 0, 0),
        metricRegistry);
    // new resource: BOOTSTRAP -> STANDBY, should work
    newStateModel.onBecomeStandbyFromBootstrap(newMockMessage, null);
    assertStateCount(Arrays.asList("offline", "bootstrap", "standby", "leader"), Arrays.asList(0, 0, 1, 0),
        metricRegistry);
    // new resource: STANDBY -> LEADER, should work
    newStateModel.onBecomeLeaderFromStandby(newMockMessage, null);
    assertStateCount(Arrays.asList("offline", "bootstrap", "standby", "leader"), Arrays.asList(0, 0, 0, 1),
        metricRegistry);
    // new resource: LEADER -> STANDBY, should work
    newStateModel.onBecomeStandbyFromLeader(newMockMessage, null);
    assertStateCount(Arrays.asList("offline", "bootstrap", "standby", "leader"), Arrays.asList(0, 0, 1, 0),
        metricRegistry);
    // new resource: STANDBY -> INACTIVE, should work
    newStateModel.onBecomeInactiveFromStandby(newMockMessage, null);
    assertStateCount(Arrays.asList("offline", "standby", "inactive"), Arrays.asList(0, 0, 1), metricRegistry);
    // new resource: INACTIVE -> OFFLINE, should work
    newStateModel.onBecomeOfflineFromInactive(newMockMessage, null);
    assertStateCount(Arrays.asList("offline", "inactive"), Arrays.asList(1, 0), metricRegistry);
    // new resource: OFFLINE -> DROPPED, should work
    disabledPartitionSet.add(partitionName);
    newStateModel.onBecomeDroppedFromOffline(newMockMessage, null);
    assertStateCount(Arrays.asList("offline"), Arrays.asList(0), metricRegistry);
    assertEquals("Dropped count should be updated", 3, participantMetrics.partitionDroppedCount.getCount());
    assertTrue("Partition should be removed from disabled partition set", disabledPartitionSet.isEmpty());
    assertEquals("Mismatch in enabled partition", partitionName, enabledPartitionSet.iterator().next());
    // new resource: ERROR -> DROPPED, should work
    newStateModel.onBecomeDroppedFromError(newMockMessage, null);
    assertEquals("Dropped count should be updated", 4, participantMetrics.partitionDroppedCount.getCount());
    // new resource: ERROR -> OFFLINE (this occurs when we use Helix API to reset certain partition in ERROR state)
    newStateModel.onBecomeOfflineFromError(newMockMessage, null);
    assertStateCount(Arrays.asList("offline"), Arrays.asList(1), metricRegistry);
    // reset method
    newStateModel.reset();
    assertStateCount(Arrays.asList("offline"), Arrays.asList(1), metricRegistry);
    // call reset method again to mock the case where same partition is reset multiple times during zk disconnection or shutdown
    newStateModel.reset();
    assertStateCount(Arrays.asList("offline"), Arrays.asList(1), metricRegistry);

    MockHelixManagerFactory.overrideGetHelixManager = false;
  }
Test that {@link HelixParticipantMetrics} keeps track of partition during state transition
ambry-clustermap/src/test/java/com/github/ambry/clustermap/AmbryStateModelFactoryTest.java