public static void removeMostServicingBrokersForNamespace(
            final String assignedBundleName,
            final Set<String> candidates,
            final ConcurrentOpenHashMap<String, ConcurrentOpenHashMap<String, ConcurrentOpenHashSet<String>>>
                    brokerToNamespaceToBundleRange) {
        if (candidates.isEmpty()) {
            return;
        }

        final String namespaceName = getNamespaceNameFromBundleName(assignedBundleName);
        int leastBundles = Integer.MAX_VALUE;

        for (final String broker : candidates) {
            int bundles = (int) brokerToNamespaceToBundleRange
                    .computeIfAbsent(broker,
                            k -> ConcurrentOpenHashMap.<String,
                                    ConcurrentOpenHashSet<String>>newBuilder().build())
                    .computeIfAbsent(namespaceName,
                            k -> ConcurrentOpenHashSet.<String>newBuilder().build())
                    .size();
            leastBundles = Math.min(leastBundles, bundles);
            if (leastBundles == 0) {
                break;
            }
        }

        // Since `brokerToNamespaceToBundleRange` can be updated by other threads,
        // `leastBundles` may differ from the actual value.

        final int finalLeastBundles = leastBundles;
        candidates.removeIf(
                broker -> brokerToNamespaceToBundleRange.computeIfAbsent(broker,
                        k -> ConcurrentOpenHashMap.<String,
                                ConcurrentOpenHashSet<String>>newBuilder().build())
                        .computeIfAbsent(namespaceName,
                                k -> ConcurrentOpenHashSet.<String>newBuilder().build())
                        .size() > finalLeastBundles);
    }
@Test
    public void testRemoveMostServicingBrokersForNamespace() {
        String namespace = "tenant1/ns1";
        String assignedBundle = namespace + "/0x00000000_0x40000000";

        Set<String> candidates = new HashSet<>();
        ConcurrentOpenHashMap<String, ConcurrentOpenHashMap<String, ConcurrentOpenHashSet<String>>> map =
                ConcurrentOpenHashMap.<String,
                        ConcurrentOpenHashMap<String, ConcurrentOpenHashSet<String>>>newBuilder()
                        .build();
        LoadManagerShared.removeMostServicingBrokersForNamespace(assignedBundle, candidates, map);
        Assert.assertEquals(candidates.size(), 0);

        candidates = Sets.newHashSet("broker1");
        LoadManagerShared.removeMostServicingBrokersForNamespace(assignedBundle, candidates, map);
        Assert.assertEquals(candidates.size(), 1);
        Assert.assertTrue(candidates.contains("broker1"));

        candidates = Sets.newHashSet("broker1");
        fillBrokerToNamespaceToBundleMap(map, "broker1", namespace, "0x40000000_0x80000000");
        LoadManagerShared.removeMostServicingBrokersForNamespace(assignedBundle, candidates, map);
        Assert.assertEquals(candidates.size(), 1);
        Assert.assertTrue(candidates.contains("broker1"));

        candidates = Sets.newHashSet("broker1", "broker2");
        LoadManagerShared.removeMostServicingBrokersForNamespace(assignedBundle, candidates, map);
        Assert.assertEquals(candidates.size(), 1);
        Assert.assertTrue(candidates.contains("broker2"));

        candidates = Sets.newHashSet("broker1", "broker2");
        fillBrokerToNamespaceToBundleMap(map, "broker2", namespace, "0x80000000_0xc0000000");
        LoadManagerShared.removeMostServicingBrokersForNamespace(assignedBundle, candidates, map);
        Assert.assertEquals(candidates.size(), 2);
        Assert.assertTrue(candidates.contains("broker1"));
        Assert.assertTrue(candidates.contains("broker2"));

        candidates = Sets.newHashSet("broker1", "broker2");
        fillBrokerToNamespaceToBundleMap(map, "broker2", namespace, "0xc0000000_0xd0000000");
        LoadManagerShared.removeMostServicingBrokersForNamespace(assignedBundle, candidates, map);
        Assert.assertEquals(candidates.size(), 1);
        Assert.assertTrue(candidates.contains("broker1"));

        candidates = Sets.newHashSet("broker1", "broker2", "broker3");
        fillBrokerToNamespaceToBundleMap(map, "broker3", namespace, "0xd0000000_0xffffffff");
        LoadManagerShared.removeMostServicingBrokersForNamespace(assignedBundle, candidates, map);
        Assert.assertEquals(candidates.size(), 2);
        Assert.assertTrue(candidates.contains("broker1"));
        Assert.assertTrue(candidates.contains("broker3"));
    }