private Long ensureLeaderConfig(long maxOffset, WorkerCoordinator coordinator) {
        // If this leader is behind some other members, we can't do assignment
        if (coordinator.configSnapshot().offset() < maxOffset) {
            // We might be able to take a new snapshot to catch up immediately and avoid another round of syncing here.
            // Alternatively, if this node has already passed the maximum reported by any other member of the group, it
            // is also safe to use this newer state.
            ClusterConfigState updatedSnapshot = coordinator.configFreshSnapshot();
            if (updatedSnapshot.offset() < maxOffset) {
                log.info("Was selected to perform assignments, but do not have latest config found in sync request. " +
                        "Returning an empty configuration to trigger re-sync.");
                return null;
            } else {
                coordinator.configSnapshot(updatedSnapshot);
                return updatedSnapshot.offset();
            }
        }
        return maxOffset;
    }
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