private void assertCompleteAllocation() {
        List<String> allAssignedConnectors = ConnectUtils.combineCollections(memberAssignments.values(), ConnectorsAndTasks::connectors);
        assertEquals(connectors.keySet(),
            new HashSet<>(allAssignedConnectors),
            "The set of connectors assigned across the cluster does not match the set of connectors in the config topic");

        Map<String, List<ConnectorTaskId>> allAssignedTasks = ConnectUtils.combineCollections(memberAssignments.values(), ConnectorsAndTasks::tasks)
                .stream()
                .collect(Collectors.groupingBy(ConnectorTaskId::connector, Collectors.toList()));

        connectors.forEach((connector, taskCount) -> {
            Set<ConnectorTaskId> expectedTasks = IntStream.range(0, taskCount)
                    .mapToObj(i -> new ConnectorTaskId(connector, i))
                    .collect(Collectors.toSet());
            assertEquals(expectedTasks,
                new HashSet<>(allAssignedTasks.get(connector)),
                "The set of tasks assigned across the cluster for connector " + connector + " does not match the set of tasks in the config topic");
        });
    }
Ensure that no connectors or tasks that were already assigned during the previous round are newly assigned in this round,
and that each newly-assigned connector and task is only assigned to a single worker.
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignorTest.java