YeonCheolGit commented on code in PR #12019: URL: https://github.com/apache/kafka/pull/12019#discussion_r847446537
########## connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignor.java: ########## @@ -200,283 +229,149 @@ private Long ensureLeaderConfig(long maxOffset, WorkerCoordinator coordinator) { previousAssignment = activeAssignments; canRevoke = true; } - previousRevocation.connectors().clear(); - previousRevocation.tasks().clear(); + previousRevocation = ConnectorsAndTasks.EMPTY; } - // Derived set: The set of deleted connectors-and-tasks is a derived set from the set - // difference of previous - configured - ConnectorsAndTasks deleted = diff(previousAssignment, configured); - log.debug("Deleted assignments: {}", deleted); - - // Derived set: The set of remaining active connectors-and-tasks is a derived set from the - // set difference of active - deleted - ConnectorsAndTasks remainingActive = diff(activeAssignments, deleted); - log.debug("Remaining (excluding deleted) active assignments: {}", remainingActive); - - // Derived set: The set of lost or unaccounted connectors-and-tasks is a derived set from - // the set difference of previous - active - deleted - ConnectorsAndTasks lostAssignments = diff(previousAssignment, activeAssignments, deleted); - log.debug("Lost assignments: {}", lostAssignments); - - // Derived set: The set of new connectors-and-tasks is a derived set from the set - // difference of configured - previous - active - ConnectorsAndTasks newSubmissions = diff(configured, previousAssignment, activeAssignments); - log.debug("New assignments: {}", newSubmissions); + // The connectors and tasks that have been deleted since the last rebalance + final ConnectorsAndTasks deleted = ConnectorsAndTasks.diff(previousAssignment, configured); + log.trace("Deleted assignments: {}", deleted); - // A collection of the complete assignment - List<WorkerLoad> completeWorkerAssignment = workerAssignment(memberConfigs, ConnectorsAndTasks.EMPTY); - log.debug("Complete (ignoring deletions) worker assignments: {}", completeWorkerAssignment); + // The connectors and tasks that are currently running on more than one worker each + final ConnectorsAndTasks duplicated = duplicated(memberAssignments); + log.trace("Duplicated assignments: {}", duplicated); - // Per worker connector assignments without removing deleted connectors yet - Map<String, Collection<String>> connectorAssignments = - completeWorkerAssignment.stream().collect(Collectors.toMap(WorkerLoad::worker, WorkerLoad::connectors)); - log.debug("Complete (ignoring deletions) connector assignments: {}", connectorAssignments); + // The connectors and tasks that should already be running on the cluster, but which are not included + // in the assignment reported by any workers in the cluster + final ConnectorsAndTasks lostAssignments = ConnectorsAndTasks.diff(previousAssignment, activeAssignments, deleted); + log.trace("Lost assignments: {}", lostAssignments); - // Per worker task assignments without removing deleted connectors yet - Map<String, Collection<ConnectorTaskId>> taskAssignments = - completeWorkerAssignment.stream().collect(Collectors.toMap(WorkerLoad::worker, WorkerLoad::tasks)); - log.debug("Complete (ignoring deletions) task assignments: {}", taskAssignments); + // The connectors and tasks that have been created since the last rebalance + final ConnectorsAndTasks created = ConnectorsAndTasks.diff(configured, previousAssignment, activeAssignments); + log.trace("New assignments: {}", created); - // A collection of the current assignment excluding the connectors-and-tasks to be deleted - List<WorkerLoad> currentWorkerAssignment = workerAssignment(memberConfigs, deleted); + final Map<String, ConnectorsAndTasks.Builder> toRevoke = new HashMap<>(); - Map<String, ConnectorsAndTasks> toRevoke = computeDeleted(deleted, connectorAssignments, taskAssignments); - log.debug("Connector and task to delete assignments: {}", toRevoke); + final Map<String, ConnectorsAndTasks> deletedAndRevoked = intersection(deleted, memberAssignments); + log.trace("Deleted connectors and tasks to revoke from each worker: {}", deletedAndRevoked); + addAll(toRevoke, deletedAndRevoked); // Revoking redundant connectors/tasks if the workers have duplicate assignments - toRevoke.putAll(computeDuplicatedAssignments(memberConfigs, connectorAssignments, taskAssignments)); - log.debug("Connector and task to revoke assignments (include duplicated assignments): {}", toRevoke); + final Map<String, ConnectorsAndTasks> duplicatedAndRevoked = intersection(duplicated, memberAssignments); + log.trace("Duplicated connectors and tasks to revoke from each worker: {}", duplicatedAndRevoked); + addAll(toRevoke, duplicatedAndRevoked); - // Recompute the complete assignment excluding the deleted connectors-and-tasks - completeWorkerAssignment = workerAssignment(memberConfigs, deleted); - connectorAssignments = - completeWorkerAssignment.stream().collect(Collectors.toMap(WorkerLoad::worker, WorkerLoad::connectors)); - taskAssignments = - completeWorkerAssignment.stream().collect(Collectors.toMap(WorkerLoad::worker, WorkerLoad::tasks)); + // Compute the assignment that will be applied across the cluster after this round of rebalance + // Later on, new submissions and lost-and-reassigned connectors and tasks will be added to these assignments, + // and load-balancing revocations will be removed from them. + final List<WorkerLoad> nextWorkerAssignment = workerLoads(memberAssignments, ConnectorsAndTasks.combine(deleted, duplicated)); - handleLostAssignments(lostAssignments, newSubmissions, completeWorkerAssignment, memberConfigs); + final ConnectorsAndTasks.Builder lostAssignmentsToReassign = ConnectorsAndTasks.builder(); + handleLostAssignments(lostAssignments, lostAssignmentsToReassign, nextWorkerAssignment); // Do not revoke resources for re-assignment while a delayed rebalance is active - // Also we do not revoke in two consecutive rebalances by the same leader - canRevoke = delay == 0 && canRevoke; + // Also we do not revoke in two consecutive rebalances by the same leader, which + // should be fine since workers immediately rejoin the group after a rebalance + // if connectors and/or tasks were revoked from them + canRevoke = canRevoke && delay == 0; // Compute the connectors-and-tasks to be revoked for load balancing without taking into // account the deleted ones. - log.debug("Can leader revoke tasks in this assignment? {} (delay: {})", canRevoke, delay); + log.trace("Can leader revoke tasks in this assignment? {} (delay: {})", canRevoke, delay); if (canRevoke) { - Map<String, ConnectorsAndTasks> toExplicitlyRevoke = - performTaskRevocation(activeAssignments, currentWorkerAssignment); - - log.debug("Connector and task to revoke assignments: {}", toRevoke); - - toExplicitlyRevoke.forEach( - (worker, assignment) -> { - ConnectorsAndTasks existing = toRevoke.computeIfAbsent( - worker, - v -> new ConnectorsAndTasks.Builder().build()); - existing.connectors().addAll(assignment.connectors()); - existing.tasks().addAll(assignment.tasks()); - } - ); - canRevoke = toExplicitlyRevoke.size() == 0; + Map<String, ConnectorsAndTasks> loadBalancingRevocations = + performLoadBalancingRevocations(configured, nextWorkerAssignment); + + log.trace("Load-balancing revocations for each worker: {}", loadBalancingRevocations); + addAll(toRevoke, loadBalancingRevocations); + canRevoke = ConnectorsAndTasks.combine(loadBalancingRevocations.values()).isEmpty(); } else { canRevoke = delay == 0; } - assignConnectors(completeWorkerAssignment, newSubmissions.connectors()); - assignTasks(completeWorkerAssignment, newSubmissions.tasks()); - log.debug("Current complete assignments: {}", currentWorkerAssignment); - log.debug("New complete assignments: {}", completeWorkerAssignment); + final ConnectorsAndTasks toAssign = ConnectorsAndTasks.combine(created, lostAssignmentsToReassign.build()); + assignConnectors(nextWorkerAssignment, toAssign.connectors()); + assignTasks(nextWorkerAssignment, toAssign.tasks()); - Map<String, Collection<String>> currentConnectorAssignments = - currentWorkerAssignment.stream().collect(Collectors.toMap(WorkerLoad::worker, WorkerLoad::connectors)); - Map<String, Collection<ConnectorTaskId>> currentTaskAssignments = - currentWorkerAssignment.stream().collect(Collectors.toMap(WorkerLoad::worker, WorkerLoad::tasks)); - Map<String, Collection<String>> incrementalConnectorAssignments = - diff(connectorAssignments, currentConnectorAssignments); - Map<String, Collection<ConnectorTaskId>> incrementalTaskAssignments = - diff(taskAssignments, currentTaskAssignments); + // The complete set of connectors and tasks that will be running on each worker after this rebalance + final Map<String, ConnectorsAndTasks> nextAssignments = assignments(nextWorkerAssignment); - log.debug("Incremental connector assignments: {}", incrementalConnectorAssignments); - log.debug("Incremental task assignments: {}", incrementalTaskAssignments); + log.debug("Current complete assignments: {}", memberAssignments); + log.debug("New complete assignments: {}", nextAssignments); Review Comment: @C0urante Thank you for the reply:) Maybe keeping `next` is looking better for me. Because if we change `new` to `next`, we need to change all variable names and consider huge side effects. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org