private boolean assertConnectorAndTasksAreUniqueAndBalanced() {
        try {
            Map<String, Collection<String>> connectors = new HashMap<>();
            Map<String, Collection<String>> tasks = new HashMap<>();
            for (String connector : connect.connectors()) {
                ConnectorStateInfo info = connect.connectorStatus(connector);
                connectors.computeIfAbsent(info.connector().workerId(), k -> new ArrayList<>())
                        .add(connector);
                info.tasks().forEach(
                    t -> tasks.computeIfAbsent(t.workerId(), k -> new ArrayList<>())
                           .add(connector + "-" + t.id()));
            }

            int maxConnectors = connectors.values().stream().mapToInt(Collection::size).max().orElse(0);
            int minConnectors = connectors.values().stream().mapToInt(Collection::size).min().orElse(0);
            int maxTasks = tasks.values().stream().mapToInt(Collection::size).max().orElse(0);
            int minTasks = tasks.values().stream().mapToInt(Collection::size).min().orElse(0);

            log.debug("Connector balance: {}", formatAssignment(connectors));
            log.debug("Task balance: {}", formatAssignment(tasks));

            assertNotEquals(0, maxConnectors, "Found no connectors running!");
            assertNotEquals(0, maxTasks, "Found no tasks running!");
            assertEquals(connectors.values().size(),
                    connectors.values().stream().distinct().count(),
                    "Connector assignments are not unique: " + connectors);
            assertEquals(tasks.values().size(),
                    tasks.values().stream().distinct().count(),
                    "Task assignments are not unique: " + tasks);
            assertTrue(maxConnectors - minConnectors < 2, "Connectors are imbalanced: " + formatAssignment(connectors));
            assertTrue(maxTasks - minTasks < 2, "Tasks are imbalanced: " + formatAssignment(tasks));
            return true;
        } catch (Exception e) {
            log.error("Could not check connector state info.", e);
            return false;
        }
    }
Integration tests for incremental cooperative rebalancing between Connect workers
connect/runtime/src/test/java/org/apache/kafka/connect/integration/RebalanceSourceConnectorsIntegrationTest.java