private long findMaxMemberConfigOffset(Map<String, ExtendedWorkerState> memberConfigs,
                                           WorkerCoordinator coordinator) {
        // The new config offset is the maximum seen by any member. We always perform assignment using this offset,
        // even if some members have fallen behind. The config offset used to generate the assignment is included in
        // the response so members that have fallen behind will not use the assignment until they have caught up.
        Long maxOffset = null;
        for (Map.Entry<String, ExtendedWorkerState> stateEntry : memberConfigs.entrySet()) {
            long memberRootOffset = stateEntry.getValue().offset();
            if (maxOffset == null)
                maxOffset = memberRootOffset;
            else
                maxOffset = Math.max(maxOffset, memberRootOffset);
        }

        log.debug("Max config offset root: {}, local snapshot config offsets root: {}",
                  maxOffset, coordinator.configSnapshot().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