@Override
  public List<String> assignSegment(String segmentName, Map<String, Map<String, String>> currentAssignment,
      Map<InstancePartitionsType, InstancePartitions> instancePartitionsMap) {
    Preconditions.checkState(instancePartitionsMap.size() == 1, "One instance partition type should be provided");
    InstancePartitions instancePartitions = instancePartitionsMap.get(InstancePartitionsType.CONSUMING);
    Preconditions.checkState(instancePartitions != null, "Failed to find CONSUMING instance partitions for table: %s",
        _tableNameWithType);
    _logger.info("Assigning segment: {} with instance partitions: {} for table: {}", segmentName, instancePartitions,
        _tableNameWithType);

    int partitionId = getPartitionId(segmentName);
    List<String> instancesAssigned = assignConsumingSegment(partitionId, instancePartitions);
    Set<String> existingAssignment = getExistingAssignment(partitionId, currentAssignment);
    // Check if the candidate assignment is consistent with existing assignment. Use existing assignment if not.
    if (existingAssignment == null) {
      _logger.info("No existing assignment from idealState, using the one decided by instancePartitions");
    } else if (!isSameAssignment(existingAssignment, instancesAssigned)) {
      _logger.warn("Assignment: {} is inconsistent with idealState: {}, using the one from idealState",
          instancesAssigned, existingAssignment);
      instancesAssigned = new ArrayList<>(existingAssignment);
      if (_controllerMetrics != null) {
        _controllerMetrics.addMeteredTableValue(_tableNameWithType,
            ControllerMeter.CONTROLLER_REALTIME_TABLE_SEGMENT_ASSIGNMENT_MISMATCH, 1L);
      }
    }
    _logger.info("Assigned segment: {} to instances: {} for table: {}", segmentName, instancesAssigned,
        _tableNameWithType);
    return instancesAssigned;
  }
@Test
  public void testAssignSegment() {
    assertTrue(_segmentAssignment instanceof StrictRealtimeSegmentAssignment);
    Map<InstancePartitionsType, InstancePartitions> onlyConsumingInstancePartitionMap =
        ImmutableMap.of(InstancePartitionsType.CONSUMING, _instancePartitionsMap.get(InstancePartitionsType.CONSUMING));
    int numInstancesPerReplicaGroup = NUM_CONSUMING_INSTANCES / NUM_REPLICAS;
    Map<String, Map<String, String>> currentAssignment = new TreeMap<>();
    // Add segments for partition 0/1/2, but add no segment for partition 3.
    List<String> instancesAssigned;
    boolean consistent;
    for (int segmentId = 0; segmentId < 3; segmentId++) {
      String segmentName = _segments.get(segmentId);
      instancesAssigned =
          _segmentAssignment.assignSegment(segmentName, currentAssignment, onlyConsumingInstancePartitionMap);
      assertEquals(instancesAssigned.size(), NUM_REPLICAS);
      // Segment 0 (partition 0) should be assigned to instance 0, 3, 6
      // Segment 1 (partition 1) should be assigned to instance 1, 4, 7
      // Segment 2 (partition 2) should be assigned to instance 2, 5, 8
      // Following segments are assigned to those instances if continue to use the same instancePartition
      // Segment 3 (partition 3) should be assigned to instance 0, 3, 6
      // Segment 4 (partition 0) should be assigned to instance 0, 3, 6
      // Segment 5 (partition 1) should be assigned to instance 1, 4, 7
      // ...
      for (int replicaGroupId = 0; replicaGroupId < NUM_REPLICAS; replicaGroupId++) {
        int partitionId = segmentId % NUM_PARTITIONS;
        int expectedAssignedInstanceId =
            partitionId % numInstancesPerReplicaGroup + replicaGroupId * numInstancesPerReplicaGroup;
        assertEquals(instancesAssigned.get(replicaGroupId), CONSUMING_INSTANCES.get(expectedAssignedInstanceId));
      }
      addToAssignment(currentAssignment, segmentId, instancesAssigned);
    }
    // Use new instancePartition to assign the new segments below.
    ImmutableMap<InstancePartitionsType, InstancePartitions> newConsumingInstancePartitionMap =
        ImmutableMap.of(InstancePartitionsType.CONSUMING, _newConsumingInstancePartitions);

    // No existing segments for partition 3, so use the assignment decided by new instancePartition.
    // So segment 3 (partition 3) should be assigned to instance new_0, new_3, new_6
    int segmentId = 3;
    String segmentName = _segments.get(segmentId);
    instancesAssigned =
        _segmentAssignment.assignSegment(segmentName, currentAssignment, newConsumingInstancePartitionMap);
    assertEquals(instancesAssigned,
        Arrays.asList("new_consumingInstance_0", "new_consumingInstance_3", "new_consumingInstance_6"));
    addToAssignment(currentAssignment, segmentId, instancesAssigned);

    // Use existing assignment for partition 0/1/2, instead of the one decided by new instancePartition.
    for (segmentId = 4; segmentId < 7; segmentId++) {
      segmentName = _segments.get(segmentId);
      instancesAssigned =
          _segmentAssignment.assignSegment(segmentName, currentAssignment, newConsumingInstancePartitionMap);
      assertEquals(instancesAssigned.size(), NUM_REPLICAS);

      // Those segments are assigned according to the assignment from idealState, instead of using new_xxx instances
      // Segment 4 (partition 0) should be assigned to instance 0, 3, 6
      // Segment 5 (partition 1) should be assigned to instance 1, 4, 7
      // Segment 6 (partition 2) should be assigned to instance 2, 5, 8
      for (int replicaGroupId = 0; replicaGroupId < NUM_REPLICAS; replicaGroupId++) {
        int partitionId = segmentId % NUM_PARTITIONS;
        int expectedAssignedInstanceId =
            partitionId % numInstancesPerReplicaGroup + replicaGroupId * numInstancesPerReplicaGroup;
        assertEquals(instancesAssigned.get(replicaGroupId), CONSUMING_INSTANCES.get(expectedAssignedInstanceId));
      }
      addToAssignment(currentAssignment, segmentId, instancesAssigned);
    }
  }