@Test
  public void clusterMapOverrideEnabledAndDisabledTest() throws Exception {
    assumeTrue(!useComposite && listenCrossColo);

    // Get the writable partitions in OverrideMap
    Set<String> writableInOverrideMap = new HashSet<>();
    for (Map.Entry<String, Map<String, String>> entry : partitionOverrideMap.entrySet()) {
      if (!entry.getValue().get(ClusterMapUtils.PARTITION_STATE).equals(READ_ONLY_STR)) {
        writableInOverrideMap.add(entry.getKey());
      }
    }
    // Get the writable partitions in InstanceConfig(PartitionLayout)
    List<PartitionId> writableInLayout = testPartitionLayout.getPartitionLayout().getWritablePartitions(null);
    Set<String> writableInInstanceConfig = new HashSet<>();
    writableInLayout.forEach(k -> writableInInstanceConfig.add(k.toPathString()));

    if (overrideEnabled) {
      // Verify clustermap uses partition override for initialization
      Set<String> writableInClusterManager = getWritablePartitions().getSecond();
      assertEquals("Mismatch in writable partitions during initialization", writableInOverrideMap,
          writableInClusterManager);
      // Ensure clustermap ignores the InstanceConfig when override is enabled.
      assertFalse(
          "Writable partitions in ClusterManager should not equal to those in InstanceConfigs when override is enabled",
          writableInClusterManager.equals(writableInInstanceConfig));

      // Verify writable partitions in clustermap remain unchanged when instanceConfig changes in Helix cluster
      AmbryPartition partition = (AmbryPartition) clusterManager.getWritablePartitionIds(null).get(0);
      List<String> instances = helixCluster.getInstancesForPartition((partition.toPathString()));
      Counter instanceTriggerCounter =
          ((HelixClusterManager) clusterManager).helixClusterManagerMetrics.dataNodeConfigChangeTriggerCount;
      long countVal = instanceTriggerCounter.getCount();
      helixCluster.setReplicaState(partition, instances.get(0), ReplicaStateType.SealedState, true, false);
      assertEquals("Mismatch in instanceTriggerCounter", countVal + 1, instanceTriggerCounter.getCount());
      writableInClusterManager = getWritablePartitions().getSecond();
      assertEquals("Mismatch in writable partitions when instanceConfig changes", writableInOverrideMap,
          writableInClusterManager);

      // Verify the partition state could be changed if this partition is not in partition override map.
      // Following test re-initializes clusterManager with new partitionLayout and then triggers instanceConfig change on new added partition
      testPartitionLayout.addNewPartitions(1, DEFAULT_PARTITION_CLASS, PartitionState.READ_WRITE, helixDcs[0]);
      Utils.writeJsonObjectToFile(testPartitionLayout.getPartitionLayout().toJSONObject(), partitionLayoutPath);
      helixCluster.upgradeWithNewPartitionLayout(partitionLayoutPath,
          HelixBootstrapUpgradeUtil.HelixAdminOperation.BootstrapCluster);
      clusterManager.close();
      Map<String, ZNRecord> znRecordMap = new HashMap<>();
      znRecordMap.put(PARTITION_OVERRIDE_ZNODE_PATH, znRecord);
      MockHelixManagerFactory helixManagerFactory =
          new MockHelixManagerFactory(helixCluster, znRecordMap, null, useAggregatedView);
      HelixClusterManager clusterManager =
          new HelixClusterManager(clusterMapConfig, selfInstanceName, helixManagerFactory, new MetricRegistry());
      // Ensure the new RW partition is added
      assertEquals("Mismatch in writable partitions when instanceConfig changes", writableInOverrideMap.size() + 1,
          clusterManager.getWritablePartitionIds(null).size());
      // Find out the new added partition which is not in partition override map
      for (PartitionId partitionId : clusterManager.getAllPartitionIds(null)) {
        if (partitionId.toPathString().equals(String.valueOf(testPartitionLayout.getPartitionCount() - 1))) {
          partition = (AmbryPartition) partitionId;
        }
      }
      instances = helixCluster.getInstancesForPartition((partition.toPathString()));
      // Change the replica from RW to RO, which triggers instanceConfig change
      helixCluster.setReplicaState(partition, instances.get(0), ReplicaStateType.SealedState, true, false);
      // Ensure the partition state becomes Read_Only
      assertFalse("If any one replica is SEALED, the whole partition should be SEALED",
          clusterManager.getWritablePartitionIds(null).contains(partition));
      assertEquals("If any one replica is SEALED, the whole partition should be SEALED", PartitionState.READ_ONLY,
          partition.getPartitionState());
      clusterManager.close();
    } else {
      // Verify clustermap uses instanceConfig for initialization when override map is non-empty but disabled.
      Set<String> writableInClusterManager = getWritablePartitions().getSecond();
      assertEquals("Mismatch in writable partitions during initialization", writableInInstanceConfig,
          writableInClusterManager);
      // Ensure clustermap ignores partition override map when override is disabled.
      assertFalse(
          "Writable partitions in ClusterManager should not equal to those in OverrideMap when override is disabled",
          writableInClusterManager.equals(writableInOverrideMap));

      // Verify partition state in clustermap is changed when instanceConfig changes in Helix cluster.
      // This is to ensure partition override doesn't take any effect when it is disabled.
      AmbryPartition partition = (AmbryPartition) clusterManager.getWritablePartitionIds(null).get(0);
      List<String> instances = helixCluster.getInstancesForPartition((partition.toPathString()));
      helixCluster.setReplicaState(partition, instances.get(0), ReplicaStateType.SealedState, true, false);
      assertFalse("If any one replica is SEALED, the whole partition should be SEALED",
          clusterManager.getWritablePartitionIds(null).contains(partition));
      assertEquals("If any one replica is SEALED, the whole partition should be SEALED", PartitionState.READ_ONLY,
          partition.getPartitionState());
      // Ensure that after instanceConfig changes, the writable partitions in clusterManager match those in InstanceConfig
      writableInInstanceConfig.remove(partition.toPathString());
      writableInClusterManager = getWritablePartitions().getSecond();
      assertEquals("Mismatch in writable partitions during initialization", writableInInstanceConfig,
          writableInClusterManager);
    }
  }
Test that ClusterManger will use seal state in PartitionOverride/InstanceConfig when {@link ClusterMapConfig#clusterMapEnablePartitionOverride}
is enabled/disabled. This test verifies that InstanceConfig changes won't affect any seal state of partition if clusterMapEnablePartitionOverride
is enabled. It also tests seal state can be dynamically changed by InstanceConfig change when PartitionOverride is
non-empty but disabled.
ambry-clustermap/src/test/java/com/github/ambry/clustermap/HelixClusterManagerTest.java