Repository: apache/kafka
File: connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/EagerAssignor.java (Java)

Code before:
        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);
    }

    private Map<String, ByteBuffer> fillAssignmentsAndSerialize(Collection<String> members,
                                                                short error,
                                                                String leaderId,
                                                                String leaderUrl,
                                                                long maxOffset,
                                                                Map<String, Collection<String>> connectorAssignments,
                                                                Map<String, Collection<ConnectorTaskId>> taskAssignments) {

        Map<String, ByteBuffer> groupAssignment = new HashMap<>();
        for (String member : members) {
            Collection<String> connectors = connectorAssignments.get(member);
            if (connectors == null) {
                connectors = List.of();
            }
            Collection<ConnectorTaskId> tasks = taskAssignments.get(member);
            if (tasks == null) {
                tasks = List.of();
            }
            Assignment assignment = new Assignment(error, leaderId, leaderUrl, maxOffset, connectors, tasks);
            log.debug("Assignment: {} -> {}", member, assignment);
            groupAssignment.put(member, ConnectProtocol.serializeAssignment(assignment));
        }
        log.debug("Finished assignment");
        return groupAssignment;
    }

    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;

Review comment (suggestion):
```java
            Collection<String> connectors = connectorAssignments.getOrDefault(member, List.of());
            Collection<ConnectorTaskId> tasks = taskAssignments.getOrDefault(member, List.of());
```

Code after:
        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);
    }

    private Map<String, ByteBuffer> fillAssignmentsAndSerialize(Collection<String> members,
                                                                short error,
                                                                String leaderId,
                                                                String leaderUrl,
                                                                long maxOffset,
                                                                Map<String, Collection<String>> connectorAssignments,
                                                                Map<String, Collection<ConnectorTaskId>> taskAssignments) {

        Map<String, ByteBuffer> groupAssignment = new HashMap<>();
        for (String member : members) {
            Collection<String> connectors = connectorAssignments.getOrDefault(member, List.of());
            if (connectors == null) {
                connectors = List.of();
            }
            Collection<ConnectorTaskId> tasks = taskAssignments.getOrDefault(member, List.of());
            if (tasks == null) {
                tasks = List.of();
            }
            Assignment assignment = new Assignment(error, leaderId, leaderUrl, maxOffset, connectors, tasks);
            log.debug("Assignment: {} -> {}", member, assignment);
            groupAssignment.put(member, ConnectProtocol.serializeAssignment(assignment));
        }
        log.debug("Finished assignment");
        return groupAssignment;
    }

    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;