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 the number of racks in cluster. Hence, the same rack"
                   + " will contain more than one replicas from the same partition.", 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));
        }
      }
    });
  }
@Test
  public void testPopulateRackInfoForReplicationFactorChange() {
    Map<String, List<Integer>> brokersByRack = new HashMap<>();
    Map<Integer, String> rackByBroker = new HashMap<>();
    // Expected: RuntimeException if replication factor (RF) is more than the number of brokers.
    assertThrows(RuntimeException.class, () -> populateRackInfoForReplicationFactorChange(Collections.singletonMap((short) (NODES.length + 1),
                                                                                                                   Collections.singleton(TOPIC)),
                                                                                          CLUSTER,
                                                                                          false,
                                                                                          brokersByRack,
                                                                                          rackByBroker));

    // Expected: RuntimeException if RF is more than the number of racks and rack-awareness check is not skipped.
    assertThrows(RuntimeException.class, () -> populateRackInfoForReplicationFactorChange(Collections.singletonMap((short) NODES.length,
                                                                                                                   Collections.singleton(TOPIC)),
                                                                                          CLUSTER,
                                                                                          false,
                                                                                          brokersByRack,
                                                                                          rackByBroker));

    // Expected: No failures if RF is more than the number of racks -- but less than the total number of brokers, and
    // rack-awareness check is skipped.
    populateRackInfoForReplicationFactorChange(Collections.singletonMap((short) NODES.length, Collections.singleton(TOPIC)),
                                               CLUSTER,
                                               true,
                                               brokersByRack,
                                               rackByBroker);
    assertEquals(2, brokersByRack.size());
    assertEquals(NODES.length, rackByBroker.size());
  }