@Test public void testPopulateRackInfoForReplicationFactorChange() { Map<String, List<Integer>> brokersByRack = new HashMap<>(); Map<Integer, String> rackByBroker = new HashMap<>(); assertThrows(RuntimeException.class, () -> populateRackInfoForReplicationFactorChange(Collections.singletonMap((short) (NODES.length + 1), Collections.singleton(TOPIC)), CLUSTER, false, brokersByRack, rackByBroker)); assertThrows(RuntimeException.class, () -> populateRackInfoForReplicationFactorChange(Collections.singletonMap((short) NODES.length, Collections.singleton(TOPIC)), CLUSTER, false, brokersByRack, rackByBroker)); populateRackInfoForReplicationFactorChange(Collections.singletonMap((short) NODES.length, Collections.singleton(TOPIC)), CLUSTER, true, brokersByRack, rackByBroker); assertEquals(2, brokersByRack.size()); assertEquals(NODES.length, rackByBroker.size()); }
public static void populateRackInfoForReplicationFactorChange(Map<Short, Set<String>> topicsByReplicationFactor, Cluster cluster, boolean skipTopicRackAwarenessCheck, Map<String, List<Integer>> brokersByRack, Map<Integer, String> rackByBroker) { for (Node node : cluster.nodes()) { String rack = getRackHandleNull(node); brokersByRack.putIfAbsent(rack, new ArrayList<>()); brokersByRack.get(rack).add(node.id()); rackByBroker.put(node.id(), rack); } topicsByReplicationFactor.forEach((replicationFactor, topics) -> { if (replicationFactor > rackByBroker.size()) { throw new RuntimeException(String.format("Unable to change replication factor (RF) of topics %s to %d since there are only %d " + "alive brokers in the cluster. Requested RF cannot be more than number of alive brokers.", topics, replicationFactor, rackByBroker.size())); } else if (replicationFactor > brokersByRack.size()) { if (skipTopicRackAwarenessCheck) { LOG.info("Target replication factor for topics {} is {}, which is larger than number of racks in cluster. Rack-awareness " + "property will be violated to add new replicas.", topics, replicationFactor); } else { throw new RuntimeException(String.format("Unable to change replication factor of topics %s to %d since there are only %d " + "racks in the cluster, to skip the rack-awareness check, set %s to true in the request.", topics, replicationFactor, brokersByRack.size(), ParameterUtils.SKIP_RACK_AWARENESS_CHECK_PARAM)); } } }); }
RunnableUtils { public static void populateRackInfoForReplicationFactorChange(Map<Short, Set<String>> topicsByReplicationFactor, Cluster cluster, boolean skipTopicRackAwarenessCheck, Map<String, List<Integer>> brokersByRack, Map<Integer, String> rackByBroker) { for (Node node : cluster.nodes()) { String rack = getRackHandleNull(node); brokersByRack.putIfAbsent(rack, new ArrayList<>()); brokersByRack.get(rack).add(node.id()); rackByBroker.put(node.id(), rack); } topicsByReplicationFactor.forEach((replicationFactor, topics) -> { if (replicationFactor > rackByBroker.size()) { throw new RuntimeException(String.format("Unable to change replication factor (RF) of topics %s to %d since there are only %d " + "alive brokers in the cluster. Requested RF cannot be more than number of alive brokers.", topics, replicationFactor, rackByBroker.size())); } else if (replicationFactor > brokersByRack.size()) { if (skipTopicRackAwarenessCheck) { LOG.info("Target replication factor for topics {} is {}, which is larger than number of racks in cluster. Rack-awareness " + "property will be violated to add new replicas.", topics, replicationFactor); } else { throw new RuntimeException(String.format("Unable to change replication factor of topics %s to %d since there are only %d " + "racks in the cluster, to skip the rack-awareness check, set %s to true in the request.", topics, replicationFactor, brokersByRack.size(), ParameterUtils.SKIP_RACK_AWARENESS_CHECK_PARAM)); } } }); } }
RunnableUtils { public static void populateRackInfoForReplicationFactorChange(Map<Short, Set<String>> topicsByReplicationFactor, Cluster cluster, boolean skipTopicRackAwarenessCheck, Map<String, List<Integer>> brokersByRack, Map<Integer, String> rackByBroker) { for (Node node : cluster.nodes()) { String rack = getRackHandleNull(node); brokersByRack.putIfAbsent(rack, new ArrayList<>()); brokersByRack.get(rack).add(node.id()); rackByBroker.put(node.id(), rack); } topicsByReplicationFactor.forEach((replicationFactor, topics) -> { if (replicationFactor > rackByBroker.size()) { throw new RuntimeException(String.format("Unable to change replication factor (RF) of topics %s to %d since there are only %d " + "alive brokers in the cluster. Requested RF cannot be more than number of alive brokers.", topics, replicationFactor, rackByBroker.size())); } else if (replicationFactor > brokersByRack.size()) { if (skipTopicRackAwarenessCheck) { LOG.info("Target replication factor for topics {} is {}, which is larger than number of racks in cluster. Rack-awareness " + "property will be violated to add new replicas.", topics, replicationFactor); } else { throw new RuntimeException(String.format("Unable to change replication factor of topics %s to %d since there are only %d " + "racks in the cluster, to skip the rack-awareness check, set %s to true in the request.", topics, replicationFactor, brokersByRack.size(), ParameterUtils.SKIP_RACK_AWARENESS_CHECK_PARAM)); } } }); } private  RunnableUtils(); }
RunnableUtils { public static void populateRackInfoForReplicationFactorChange(Map<Short, Set<String>> topicsByReplicationFactor, Cluster cluster, boolean skipTopicRackAwarenessCheck, Map<String, List<Integer>> brokersByRack, Map<Integer, String> rackByBroker) { for (Node node : cluster.nodes()) { String rack = getRackHandleNull(node); brokersByRack.putIfAbsent(rack, new ArrayList<>()); brokersByRack.get(rack).add(node.id()); rackByBroker.put(node.id(), rack); } topicsByReplicationFactor.forEach((replicationFactor, topics) -> { if (replicationFactor > rackByBroker.size()) { throw new RuntimeException(String.format("Unable to change replication factor (RF) of topics %s to %d since there are only %d " + "alive brokers in the cluster. Requested RF cannot be more than number of alive brokers.", topics, replicationFactor, rackByBroker.size())); } else if (replicationFactor > brokersByRack.size()) { if (skipTopicRackAwarenessCheck) { LOG.info("Target replication factor for topics {} is {}, which is larger than number of racks in cluster. Rack-awareness " + "property will be violated to add new replicas.", topics, replicationFactor); } else { throw new RuntimeException(String.format("Unable to change replication factor of topics %s to %d since there are only %d " + "racks in the cluster, to skip the rack-awareness check, set %s to true in the request.", topics, replicationFactor, brokersByRack.size(), ParameterUtils.SKIP_RACK_AWARENESS_CHECK_PARAM)); } } }); } private  RunnableUtils(); static void populateRackInfoForReplicationFactorChange(Map<Short, Set<String>> topicsByReplicationFactor,
                                                                Cluster cluster,
                                                                boolean skipTopicRackAwarenessCheck,
                                                                Map<String, List<Integer>> brokersByRack,
                                                                Map<Integer, String> rackByBroker); static Map<Short, Set<String>> topicsForReplicationFactorChange(Map<Short, Pattern> topicPatternByReplicationFactor,
                                                                         Cluster cluster); static boolean shouldRefreshClusterAndGeneration(Set<CruiseControlState.SubState> substates); static PartitionInfo partitionWithOfflineReplicas(Cluster cluster); static boolean isKafkaAssignerMode(Collection<String> goals); static void sanityCheckBrokersHavingOfflineReplicasOnBadDisks(List<String> goals, ClusterModel clusterModel); static void maybeStopOngoingExecutionToModifyAndWait(KafkaCruiseControl kafkaCruiseControl,
                                                              OperationProgress operationProgress); static OptimizationOptions computeOptimizationOptions(ClusterModel clusterModel,
                                                               boolean isTriggeredByGoalViolation,
                                                               KafkaCruiseControl kafkaCruiseControl,
                                                               Set<Integer> brokersToDrop,
                                                               boolean dryRun,
                                                               boolean excludeRecentlyDemotedBrokers,
                                                               boolean excludeRecentlyRemovedBrokers,
                                                               Pattern excludedTopicsPattern,
                                                               Set<Integer> requestedDestinationBrokerIds,
                                                               boolean onlyMoveImmigrantReplicas); static void sanityCheckOfflineReplicaPresence(ClusterModel clusterModel); }
RunnableUtils { public static void populateRackInfoForReplicationFactorChange(Map<Short, Set<String>> topicsByReplicationFactor, Cluster cluster, boolean skipTopicRackAwarenessCheck, Map<String, List<Integer>> brokersByRack, Map<Integer, String> rackByBroker) { for (Node node : cluster.nodes()) { String rack = getRackHandleNull(node); brokersByRack.putIfAbsent(rack, new ArrayList<>()); brokersByRack.get(rack).add(node.id()); rackByBroker.put(node.id(), rack); } topicsByReplicationFactor.forEach((replicationFactor, topics) -> { if (replicationFactor > rackByBroker.size()) { throw new RuntimeException(String.format("Unable to change replication factor (RF) of topics %s to %d since there are only %d " + "alive brokers in the cluster. Requested RF cannot be more than number of alive brokers.", topics, replicationFactor, rackByBroker.size())); } else if (replicationFactor > brokersByRack.size()) { if (skipTopicRackAwarenessCheck) { LOG.info("Target replication factor for topics {} is {}, which is larger than number of racks in cluster. Rack-awareness " + "property will be violated to add new replicas.", topics, replicationFactor); } else { throw new RuntimeException(String.format("Unable to change replication factor of topics %s to %d since there are only %d " + "racks in the cluster, to skip the rack-awareness check, set %s to true in the request.", topics, replicationFactor, brokersByRack.size(), ParameterUtils.SKIP_RACK_AWARENESS_CHECK_PARAM)); } } }); } private  RunnableUtils(); static void populateRackInfoForReplicationFactorChange(Map<Short, Set<String>> topicsByReplicationFactor,
                                                                Cluster cluster,
                                                                boolean skipTopicRackAwarenessCheck,
                                                                Map<String, List<Integer>> brokersByRack,
                                                                Map<Integer, String> rackByBroker); static Map<Short, Set<String>> topicsForReplicationFactorChange(Map<Short, Pattern> topicPatternByReplicationFactor,
                                                                         Cluster cluster); static boolean shouldRefreshClusterAndGeneration(Set<CruiseControlState.SubState> substates); static PartitionInfo partitionWithOfflineReplicas(Cluster cluster); static boolean isKafkaAssignerMode(Collection<String> goals); static void sanityCheckBrokersHavingOfflineReplicasOnBadDisks(List<String> goals, ClusterModel clusterModel); static void maybeStopOngoingExecutionToModifyAndWait(KafkaCruiseControl kafkaCruiseControl,
                                                              OperationProgress operationProgress); static OptimizationOptions computeOptimizationOptions(ClusterModel clusterModel,
                                                               boolean isTriggeredByGoalViolation,
                                                               KafkaCruiseControl kafkaCruiseControl,
                                                               Set<Integer> brokersToDrop,
                                                               boolean dryRun,
                                                               boolean excludeRecentlyDemotedBrokers,
                                                               boolean excludeRecentlyRemovedBrokers,
                                                               Pattern excludedTopicsPattern,
                                                               Set<Integer> requestedDestinationBrokerIds,
                                                               boolean onlyMoveImmigrantReplicas); static void sanityCheckOfflineReplicaPresence(ClusterModel clusterModel); static final boolean SELF_HEALING_DRYRUN; static final Set<Integer> SELF_HEALING_DESTINATION_BROKER_IDS; static final ReplicaMovementStrategy SELF_HEALING_REPLICA_MOVEMENT_STRATEGY; static final Pattern SELF_HEALING_EXCLUDED_TOPICS; static final Integer SELF_HEALING_CONCURRENT_MOVEMENTS; static final Long SELF_HEALING_EXECUTION_PROGRESS_CHECK_INTERVAL_MS; static final boolean SELF_HEALING_SKIP_HARD_GOAL_CHECK; static final boolean SELF_HEALING_STOP_ONGOING_EXECUTION; static final ModelCompletenessRequirements SELF_HEALING_MODEL_COMPLETENESS_REQUIREMENTS; static final boolean SELF_HEALING_SKIP_URP_DEMOTION; static final boolean SELF_HEALING_EXCLUDE_FOLLOWER_DEMOTION; static final boolean SELF_HEALING_SKIP_RACK_AWARENESS_CHECK; static final boolean SELF_HEALING_IS_TRIGGERED_BY_USER_REQUEST; }