static Map<Address, List<Shard>> assignShards(Collection<Shard> shards, Collection<Address> addresses) {
        Map<String, List<String>> assignment = addresses.stream()  // host -> [indexShard...]
                .map(Address::getHost).distinct().collect(toMap(identity(), a -> new ArrayList<>()));
        Map<String, List<String>> nodeCandidates = shards.stream()  // indexShard -> [host...]
                .collect(groupingBy(Shard::indexShard, mapping(Shard::getIp, toList())));

        // Make the assignment
        nodeCandidates.forEach((indexShard, hosts) -> hosts.stream()
                .map(assignment::get)
                .filter(Objects::nonNull)
                .min(comparingInt(List::size))
                .orElseThrow(() -> new IllegalStateException("Selected members do not contain shard '" + indexShard + "'"))
                .add(indexShard));

        // Transform the results
        Map<String, List<Address>> addressMap = addresses.stream().collect(groupingBy(Address::getHost, toList()));
        Map<String, Shard> shardMap = shards.stream().collect(toMap(s -> s.indexShard() + "@" + s.getIp(), identity()));
        return assignment.entrySet().stream()
                .flatMap(e -> {
                    List<Address> a = addressMap.get(e.getKey());
                    List<Shard> s = e.getValue().stream()
                            .map(indexShard -> shardMap.get(indexShard + "@" + e.getKey())).toList();
                    int c = (int) Math.ceil((double) s.size() / a.size());
                    return IntStream.range(0, a.size())
                            .mapToObj(i -> entry(a.get(i), List.copyOf(s.subList(i * c, Math.min((i + 1) * c, s.size())))));
                }).collect(toMap(Entry::getKey, Entry::getValue));
    }
@Test
    public void given_multipleNodeAddressesOnLocal_when_assignShards_then_shouldAssignSingleShardToEachAddress()
            throws UnknownHostException {
        List<Shard> shards = List.of(
                new Shard("elastic-index", 0, Prirep.p, 10, "STARTED", "127.0.0.1", "127.0.0.1:9200", "node1"),
                new Shard("elastic-index", 1, Prirep.p, 10, "STARTED", "127.0.0.1", "127.0.0.1:9201", "node2"),
                new Shard("elastic-index", 2, Prirep.p, 10, "STARTED", "127.0.0.1", "127.0.0.1:9202", "node3")
        );

        List<Address> addresses = List.of(
                new Address("127.0.0.1", 5701),
                new Address("127.0.0.1", 5702),
                new Address("127.0.0.1", 5703)
        );

        Map<Address, List<Shard>> assignment = ElasticSourcePMetaSupplier.assignShards(shards, addresses);

        assertThat(assignment)
                .containsOnlyKeys(addresses)
                // shards are distributed evenly
                .allSatisfy((address, shardList) -> assertThat(shardList).hasSize(1))
                // all shards are assigned
                .satisfies(a -> assertThat(a.values().stream().flatMap(List::stream))
                        .containsExactlyInAnyOrderElementsOf(shards));
    }