@Override
  public Map<String, Set<Integer>> assignBrokerSetsForUnresolvedBrokers(ClusterModel clusterModel, Map<String, Set<Integer>> existingBrokerSetMapping)
      throws BrokerSetResolutionException {
    // Sanity check to check if all brokers in data store do not match all brokers in cluster model
    Set<Broker> allMappedBrokers = existingBrokerSetMapping.values()
                                                           .stream()
                                                           .flatMap(Collection::stream)
                                                           .map(clusterModel::broker)
                                                           .filter(Objects::nonNull)
                                                           .collect(Collectors.toSet());

    Set<Broker> extraBrokersInClusterModel = new HashSet<>(clusterModel.brokers());
    extraBrokersInClusterModel.removeAll(allMappedBrokers);

    boolean extraClusterModelBrokersHaveReplicas = extraBrokersInClusterModel.stream().anyMatch(broker -> !broker.replicas().isEmpty());
    // The broker list in data store may not be atomically updated when brokers are added to the cluster
    // In this case we can ignore the brokers if they have no replicas placed
    if (!allMappedBrokers.equals(clusterModel.brokers()) && extraClusterModelBrokersHaveReplicas) {
      throw new BrokerSetResolutionException(
          String.format("All Brokers from data store %s do not match brokers in cluster model %s.", allMappedBrokers,
                        clusterModel.brokers()));
    }

    Set<Integer> unmappedEmptyBrokerIds = extraBrokersInClusterModel.stream()
                                                                    .filter(broker -> broker.replicas().isEmpty())
                                                                    .map(Broker::id)
                                                                    .collect(Collectors.toSet());

    if (!unmappedEmptyBrokerIds.isEmpty()) {
      existingBrokerSetMapping.computeIfAbsent(UNMAPPED_BROKER_SET_ID, k -> new HashSet<>(unmappedEmptyBrokerIds))
                              .addAll(unmappedEmptyBrokerIds);
    }

    return existingBrokerSetMapping;
  }
@Test
  public void testBrokerSetAssignment() {
    Map<Integer, String> brokers = Map.of(0, "", 1, "", 2, "", 3, "", 4, "", 5, "");
    Map<String, Set<Integer>> mappedBrokers = Map.of("bs1", Set.of(0, 1, 2, 3, 4), "bs2", Set.of(5));
    Map<String, Set<Integer>> mappedBrokersAfterAssignment =
        NO_OP_BROKER_SET_ASSIGNMENT_POLICY.assignBrokerSetsForUnresolvedBrokers(brokers, mappedBrokers);

    assertNotNull(mappedBrokersAfterAssignment);

    assertEquals(mappedBrokersAfterAssignment, mappedBrokers);
  }