private Map<String, ConnectorsAndTasks> performLoadBalancingRevocations(
            ConnectorsAndTasks configured,
            Collection<WorkerLoad> workers
    ) {
        if (log.isTraceEnabled()) {
            workers.forEach(wl -> log.trace(
                    "Per worker current load size; worker: {} connectors: {} tasks: {}",
                    wl.worker(), wl.connectorsSize(), wl.tasksSize()));
        }

        if (workers.stream().allMatch(WorkerLoad::isEmpty)) {
            log.trace("No load-balancing revocations required; all workers are either new "
                    + "or will have all currently-assigned connectors and tasks revoked during this round"
            );
            return Collections.emptyMap();
        }
        if (configured.isEmpty()) {
            log.trace("No load-balancing revocations required; no connectors are currently configured on this cluster");
            return Collections.emptyMap();
        }

        Map<String, ConnectorsAndTasks.Builder> result = new HashMap<>();

        Map<String, Set<String>> connectorRevocations = loadBalancingRevocations(
                "connector",
                configured.connectors().size(),
                workers,
                WorkerLoad::connectors
        );
        Map<String, Set<ConnectorTaskId>> taskRevocations = loadBalancingRevocations(
                "task",
                configured.tasks().size(),
                workers,
                WorkerLoad::tasks
        );

        connectorRevocations.forEach((worker, revoked) ->
                result.computeIfAbsent(worker, w -> new ConnectorsAndTasks.Builder()).addConnectors(revoked)
        );
        taskRevocations.forEach((worker, revoked) ->
                result.computeIfAbsent(worker, w -> new ConnectorsAndTasks.Builder()).addTasks(revoked)
        );

        return buildAll(result);
    }
Revoke connectors and tasks from each worker in the cluster until no worker is running more than it
would be with a perfectly-balanced assignment.
@param configured the set of configured connectors and tasks across the entire cluster
@param workers the workers in the cluster, whose assignments should not include any deleted or duplicated connectors or tasks
               that are already due to be revoked from the worker in this rebalance
@return which connectors and tasks should be revoked from which workers; never null, but may be empty
if no load-balancing revocations are necessary or possible
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignor.java