@Override
  public String brokerSetIdForReplica(final Replica replica, final ClusterModel clusterModel,
                                      final BrokerSetResolutionHelper brokerSetResolutionHelper) throws ReplicaToBrokerSetMappingException {
    String topicName = replica.topicPartition().topic();

    Map<String, Set<Integer>> brokersByBrokerSetId = brokerSetResolutionHelper.brokersByBrokerSetId();
    int numBrokerSets = brokersByBrokerSetId.size();

    if (numBrokerSets == 0) {
      throw new ReplicaToBrokerSetMappingException("Can not map replica to a broker set since there are no broker sets.");
    }

    // If the broker sets haven't changed and topic mapping is cached, then read from cache
    if (brokersByBrokerSetId.keySet().equals(_brokerSetIdCache) && _brokerSetIdByTopicCache.containsKey(topicName)) {
      return _brokerSetIdByTopicCache.get(topicName);
    }

    // If the broker set ids have changed then clear the caches and re-map topics since we use consistent hash ring
    if (!brokersByBrokerSetId.keySet().equals(_brokerSetIdCache)) {
      _brokerSetIdByTopicCache.clear();
      _sortedBrokerSetIdsCache.clear();
      _brokerSetIdCache = new HashSet<>(brokersByBrokerSetId.keySet());
    }

    String brokerSetId = brokerSetIdForTopic(topicName, brokersByBrokerSetId);
    _brokerSetIdByTopicCache.put(topicName, brokerSetId);
    return brokerSetId;
  }
@Test
  public void testSingleBrokerSetMappingPolicy() throws BrokerSetResolutionException, ReplicaToBrokerSetMappingException {
    ClusterModel clusterModel = DeterministicCluster.brokerSetSatisfiable1();
    Map<String, Set<Integer>> testSingleBrokerSetMapping = Collections.singletonMap("BS1", Set.of(0, 1, 2, 3, 4, 5));

    BrokerSetResolver brokerSetResolver = EasyMock.createNiceMock(BrokerSetResolver.class);
    EasyMock.expect(brokerSetResolver.brokerIdsByBrokerSetId(BrokerSetResolutionHelper.getRackIdByBrokerIdMapping(clusterModel)))
            .andReturn(testSingleBrokerSetMapping);
    EasyMock.replay(brokerSetResolver);

    BrokerSetResolutionHelper brokerSetResolutionHelper = new BrokerSetResolutionHelper(clusterModel, brokerSetResolver);

    for (Replica replica : clusterModel.leaderReplicas()) {
      assertEquals("BS1",
                   TOPIC_NAME_HASH_BROKER_SET_MAPPING_POLICY.brokerSetIdForReplica(replica, clusterModel, brokerSetResolutionHelper));
    }
  }