private <E> Map<String, Set<E>> loadBalancingRevocations(
            String allocatedResourceName,
            int totalToAllocate,
            Collection<WorkerLoad> workers,
            Function<WorkerLoad, Collection<E>> workerAllocation
    ) {
        int totalWorkers = workers.size();
        // The minimum instances of this resource that should be assigned to each worker
        int minAllocatedPerWorker = totalToAllocate / totalWorkers;
        // How many workers are going to have to be allocated exactly one extra instance
        // (since the total number to allocate may not be a perfect multiple of the number of workers)
        int workersToAllocateExtra = totalToAllocate % totalWorkers;
        // Useful function to determine exactly how many instances of the resource a given worker is currently allocated
        Function<WorkerLoad, Integer> workerAllocationSize = workerAllocation.andThen(Collection::size);

        long workersAllocatedMinimum = workers.stream()
                .map(workerAllocationSize)
                .filter(n -> n == minAllocatedPerWorker)
                .count();
        long workersAllocatedSingleExtra = workers.stream()
                .map(workerAllocationSize)
                .filter(n -> n == minAllocatedPerWorker + 1)
                .count();
        if (workersAllocatedSingleExtra == workersToAllocateExtra
                && workersAllocatedMinimum + workersAllocatedSingleExtra == totalWorkers) {
            log.trace(
                    "No load-balancing {} revocations required; the current allocations, when combined with any newly-created {}s, should be balanced",
                    allocatedResourceName,
                    allocatedResourceName
            );
            return Collections.emptyMap();
        }

        Map<String, Set<E>> result = new HashMap<>();
        // How many workers we've allocated a single extra resource instance to
        int allocatedExtras = 0;
        // Calculate how many (and which) connectors/tasks to revoke from each worker here
        for (WorkerLoad worker : workers) {
            int currentAllocationSizeForWorker = workerAllocationSize.apply(worker);
            if (currentAllocationSizeForWorker <= minAllocatedPerWorker) {
                // This worker isn't allocated more than the minimum; no need to revoke anything
                continue;
            }
            int maxAllocationForWorker;
            if (allocatedExtras < workersToAllocateExtra) {
                // We'll allocate one of the extra resource instances to this worker
                allocatedExtras++;
                if (currentAllocationSizeForWorker == minAllocatedPerWorker + 1) {
                    // If the worker's running exactly one more than the minimum, and we're allowed to
                    // allocate an extra to it, there's no need to revoke anything
                    continue;
                }
                maxAllocationForWorker = minAllocatedPerWorker + 1;
            } else {
                maxAllocationForWorker = minAllocatedPerWorker;
            }

            Set<E> revokedFromWorker = new LinkedHashSet<>();
            result.put(worker.worker(), revokedFromWorker);

            Iterator<E> currentWorkerAllocation = workerAllocation.apply(worker).iterator();
            // Revoke resources from the worker until it isn't allocated any more than it should be
            for (int numRevoked = 0; currentAllocationSizeForWorker - numRevoked > maxAllocationForWorker; numRevoked++) {
                if (!currentWorkerAllocation.hasNext()) {
                    // Should never happen, but better to log a warning and move on than die and fail the whole rebalance if it does
                    log.warn(
                            "Unexpectedly ran out of {}s to revoke from worker {} while performing load-balancing revocations; " +
                                    "worker appears to still be allocated {} instances, which is more than the intended allocation of {}",
                            allocatedResourceName,
                            worker.worker(),
                            workerAllocationSize.apply(worker),
                            maxAllocationForWorker
                    );
                    break;
                }
                E revocation = currentWorkerAllocation.next();
                revokedFromWorker.add(revocation);
            }
        }
        return 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