private Map<String, ByteBuffer> performTaskAssignment(String leaderId, long maxOffset,
                                                          Map<String, ExtendedWorkerState> memberConfigs,
                                                          WorkerCoordinator coordinator) {
        Map<String, Collection<String>> connectorAssignments = new HashMap<>();
        Map<String, Collection<ConnectorTaskId>> taskAssignments = new HashMap<>();

        // Perform round-robin task assignment. Assign all connectors and then all tasks because assigning both the
        // connector and its tasks can lead to very uneven distribution of work in some common cases (e.g. for connectors
        // that generate only 1 task each; in a cluster of 2 or an even # of nodes, only even nodes will be assigned
        // connectors and only odd nodes will be assigned tasks, but tasks are, on average, actually more resource
        // intensive than connectors).
        List<String> connectorsSorted = sorted(coordinator.configSnapshot().connectors());
        CircularIterator<String> memberIt = new CircularIterator<>(sorted(memberConfigs.keySet()));
        for (String connectorId : connectorsSorted) {
            String connectorAssignedTo = memberIt.next();
            log.trace("Assigning connector {} to {}", connectorId, connectorAssignedTo);
            Collection<String> memberConnectors = connectorAssignments.computeIfAbsent(connectorAssignedTo, k -> new ArrayList<>());
            memberConnectors.add(connectorId);
        }
        for (String connectorId : connectorsSorted) {
            for (ConnectorTaskId taskId : sorted(coordinator.configSnapshot().tasks(connectorId))) {
                String taskAssignedTo = memberIt.next();
                log.trace("Assigning task {} to {}", taskId, taskAssignedTo);
                Collection<ConnectorTaskId> memberTasks = taskAssignments.computeIfAbsent(taskAssignedTo, k -> new ArrayList<>());
                memberTasks.add(taskId);
            }
        }

        coordinator.leaderState(new LeaderState(memberConfigs, connectorAssignments, taskAssignments));

        return fillAssignmentsAndSerialize(memberConfigs.keySet(), Assignment.NO_ERROR,
                leaderId, memberConfigs.get(leaderId).url(), maxOffset, connectorAssignments, taskAssignments);
    }
An assignor that computes a unweighted round-robin distribution of connectors and tasks. The
connectors are assigned to the workers first, followed by the tasks. This is to avoid
load imbalance when several 1-task connectors are running, given that a connector is usually
more lightweight than a task.
<p>
Note that this class is NOT thread-safe.
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/EagerAssignor.java