diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignor.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignor.java index d4e9d878c5515..744a099730097 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignor.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignor.java @@ -244,6 +244,10 @@ protected Map performTaskAssignment(String leaderId, long ma Map toRevoke = computeDeleted(deleted, connectorAssignments, taskAssignments); log.debug("Connector and task to delete assignments: {}", toRevoke); + // Revoking redundant connectors/tasks if the the workers have duplicate assignments + toRevoke.putAll(computeDuplicatedAssignments(memberConfigs, connectorAssignments, taskAssignments)); + log.debug("Connector and task to revoke assignments (include duplicated assignments): {}", toRevoke); + // Recompute the complete assignment excluding the deleted connectors-and-tasks completeWorkerAssignment = workerAssignment(memberConfigs, deleted); connectorAssignments = @@ -359,6 +363,61 @@ private ConnectorsAndTasks computePreviousAssignment(Map memberConfigs) { + Set connectors = memberConfigs.entrySet().stream() + .flatMap(memberConfig -> memberConfig.getValue().assignment().connectors().stream()) + .collect(Collectors.groupingBy(Function.identity(), Collectors.counting())) + .entrySet().stream() + .filter(entry -> entry.getValue() > 1L) + .map(entry -> entry.getKey()) + .collect(Collectors.toSet()); + + Set tasks = memberConfigs.values().stream() + .flatMap(state -> state.assignment().tasks().stream()) + .collect(Collectors.groupingBy(Function.identity(), Collectors.counting())) + .entrySet().stream() + .filter(entry -> entry.getValue() > 1L) + .map(entry -> entry.getKey()) + .collect(Collectors.toSet()); + return new ConnectorsAndTasks.Builder().with(connectors, tasks).build(); + } + + private Map computeDuplicatedAssignments(Map memberConfigs, + Map> connectorAssignments, + Map> taskAssignment) { + ConnectorsAndTasks duplicatedAssignments = duplicatedAssignments(memberConfigs); + log.debug("Duplicated assignments: {}", duplicatedAssignments); + + Map toRevoke = new HashMap<>(); + if (!duplicatedAssignments.connectors().isEmpty()) { + connectorAssignments.entrySet().stream() + .forEach(entry -> { + Set duplicatedConnectors = new HashSet<>(duplicatedAssignments.connectors()); + duplicatedConnectors.retainAll(entry.getValue()); + if (!duplicatedConnectors.isEmpty()) { + toRevoke.computeIfAbsent( + entry.getKey(), + v -> new ConnectorsAndTasks.Builder().build() + ).connectors().addAll(duplicatedConnectors); + } + }); + } + if (!duplicatedAssignments.tasks().isEmpty()) { + taskAssignment.entrySet().stream() + .forEach(entry -> { + Set duplicatedTasks = new HashSet<>(duplicatedAssignments.tasks()); + duplicatedTasks.retainAll(entry.getValue()); + if (!duplicatedTasks.isEmpty()) { + toRevoke.computeIfAbsent( + entry.getKey(), + v -> new ConnectorsAndTasks.Builder().build() + ).tasks().addAll(duplicatedTasks); + } + }); + } + return toRevoke; + } + // visible for testing protected void handleLostAssignments(ConnectorsAndTasks lostAssignments, ConnectorsAndTasks newSubmissions, diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignorTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignorTest.java index 3e72ce9913df5..684da46bdadb1 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignorTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignorTest.java @@ -1089,6 +1089,148 @@ public void testLostAssignmentHandlingWhenWorkerBouncesBackButFinallyLeaves() { assertEquals(0, assignor.delay); } + @Test + public void testTaskAssignmentWhenTasksDuplicatedInWorkerAssignment() { + when(coordinator.configSnapshot()).thenReturn(configState); + doReturn(Collections.EMPTY_MAP).when(assignor).serializeAssignments(assignmentsCapture.capture()); + + // First assignment with 1 worker and 2 connectors configured but not yet assigned + assignor.performTaskAssignment(leader, offset, memberConfigs, coordinator, protocolVersion); + ++rebalanceNum; + returnedAssignments = assignmentsCapture.getValue(); + assertDelay(0, returnedAssignments); + expectedMemberConfigs = memberConfigs(leader, offset, returnedAssignments); + assertNoReassignments(memberConfigs, expectedMemberConfigs); + assertAssignment(2, 8, 0, 0, "worker1"); + + // Second assignment with a second worker with duplicate assignment joining and all connectors running on previous worker + applyAssignments(returnedAssignments); + memberConfigs = memberConfigs(leader, offset, assignments); + ExtendedAssignment duplicatedWorkerAssignment = newExpandableAssignment(); + duplicatedWorkerAssignment.connectors().addAll(newConnectors(1, 2)); + duplicatedWorkerAssignment.tasks().addAll(newTasks("connector1", 0, 4)); + memberConfigs.put("worker2", new ExtendedWorkerState(leaderUrl, offset, duplicatedWorkerAssignment)); + assignor.performTaskAssignment(leader, offset, memberConfigs, coordinator, protocolVersion); + ++rebalanceNum; + returnedAssignments = assignmentsCapture.getValue(); + assertDelay(0, returnedAssignments); + expectedMemberConfigs = memberConfigs(leader, offset, returnedAssignments); + assertNoReassignments(memberConfigs, expectedMemberConfigs); + assertAssignment(0, 0, 2, 8, "worker1", "worker2"); + + // Third assignment after revocations + applyAssignments(returnedAssignments); + memberConfigs = memberConfigs(leader, offset, assignments); + assignor.performTaskAssignment(leader, offset, memberConfigs, coordinator, protocolVersion); + ++rebalanceNum; + returnedAssignments = assignmentsCapture.getValue(); + assertDelay(0, returnedAssignments); + expectedMemberConfigs = memberConfigs(leader, offset, returnedAssignments); + assertNoReassignments(memberConfigs, expectedMemberConfigs); + assertAssignment(1, 4, 0, 2, "worker1", "worker2"); + + // fourth rebalance after revocations + applyAssignments(returnedAssignments); + memberConfigs = memberConfigs(leader, offset, assignments); + assignor.performTaskAssignment(leader, offset, memberConfigs, coordinator, protocolVersion); + ++rebalanceNum; + returnedAssignments = assignmentsCapture.getValue(); + assertDelay(0, returnedAssignments); + expectedMemberConfigs = memberConfigs(leader, offset, returnedAssignments); + assertNoReassignments(memberConfigs, expectedMemberConfigs); + assertAssignment(0, 2, 0, 0, "worker1", "worker2"); + + // Fifth rebalance should not change assignments + applyAssignments(returnedAssignments); + memberConfigs = memberConfigs(leader, offset, assignments); + assignor.performTaskAssignment(leader, offset, memberConfigs, coordinator, protocolVersion); + ++rebalanceNum; + returnedAssignments = assignmentsCapture.getValue(); + assertDelay(0, returnedAssignments); + expectedMemberConfigs = memberConfigs(leader, offset, returnedAssignments); + assertNoReassignments(memberConfigs, expectedMemberConfigs); + assertAssignment(0, 0, 0, 0, "worker1", "worker2"); + + verify(coordinator, times(rebalanceNum)).configSnapshot(); + verify(coordinator, times(rebalanceNum)).leaderState(any()); + verify(coordinator, times(2 * rebalanceNum)).generationId(); + verify(coordinator, times(rebalanceNum)).memberId(); + verify(coordinator, times(rebalanceNum)).lastCompletedGenerationId(); + } + + @Test + public void testDuplicatedAssignmentHandleWhenTheDuplicatedAssignmentsDeleted() { + when(coordinator.configSnapshot()).thenReturn(configState); + doReturn(Collections.EMPTY_MAP).when(assignor).serializeAssignments(assignmentsCapture.capture()); + + // First assignment with 1 worker and 2 connectors configured but not yet assigned + assignor.performTaskAssignment(leader, offset, memberConfigs, coordinator, protocolVersion); + ++rebalanceNum; + returnedAssignments = assignmentsCapture.getValue(); + assertDelay(0, returnedAssignments); + expectedMemberConfigs = memberConfigs(leader, offset, returnedAssignments); + assertNoReassignments(memberConfigs, expectedMemberConfigs); + assertAssignment(2, 8, 0, 0, "worker1"); + + //delete connector1 + configState = clusterConfigState(offset, 2, 1, 4); + when(coordinator.configSnapshot()).thenReturn(configState); + + // Second assignment with a second worker with duplicate assignment joining and the duplicated assignment is deleted at the same time + applyAssignments(returnedAssignments); + memberConfigs = memberConfigs(leader, offset, assignments); + ExtendedAssignment duplicatedWorkerAssignment = newExpandableAssignment(); + duplicatedWorkerAssignment.connectors().addAll(newConnectors(1, 2)); + duplicatedWorkerAssignment.tasks().addAll(newTasks("connector1", 0, 4)); + memberConfigs.put("worker2", new ExtendedWorkerState(leaderUrl, offset, duplicatedWorkerAssignment)); + assignor.performTaskAssignment(leader, offset, memberConfigs, coordinator, protocolVersion); + ++rebalanceNum; + returnedAssignments = assignmentsCapture.getValue(); + assertDelay(0, returnedAssignments); + expectedMemberConfigs = memberConfigs(leader, offset, returnedAssignments); + assertNoReassignments(memberConfigs, expectedMemberConfigs); + assertAssignment(0, 0, 2, 8, "worker1", "worker2"); + + // Third assignment after revocations + applyAssignments(returnedAssignments); + memberConfigs = memberConfigs(leader, offset, assignments); + assignor.performTaskAssignment(leader, offset, memberConfigs, coordinator, protocolVersion); + ++rebalanceNum; + returnedAssignments = assignmentsCapture.getValue(); + assertDelay(0, returnedAssignments); + expectedMemberConfigs = memberConfigs(leader, offset, returnedAssignments); + assertNoReassignments(memberConfigs, expectedMemberConfigs); + assertAssignment(0, 0, 0, 2, "worker1", "worker2"); + + // fourth rebalance after revocations + applyAssignments(returnedAssignments); + memberConfigs = memberConfigs(leader, offset, assignments); + assignor.performTaskAssignment(leader, offset, memberConfigs, coordinator, protocolVersion); + ++rebalanceNum; + returnedAssignments = assignmentsCapture.getValue(); + assertDelay(0, returnedAssignments); + expectedMemberConfigs = memberConfigs(leader, offset, returnedAssignments); + assertNoReassignments(memberConfigs, expectedMemberConfigs); + assertAssignment(0, 2, 0, 0, "worker1", "worker2"); + + // Fifth rebalance should not change assignments + applyAssignments(returnedAssignments); + memberConfigs = memberConfigs(leader, offset, assignments); + assignor.performTaskAssignment(leader, offset, memberConfigs, coordinator, protocolVersion); + ++rebalanceNum; + returnedAssignments = assignmentsCapture.getValue(); + assertDelay(0, returnedAssignments); + expectedMemberConfigs = memberConfigs(leader, offset, returnedAssignments); + assertNoReassignments(memberConfigs, expectedMemberConfigs); + assertAssignment(0, 0, 0, 0, "worker1", "worker2"); + + verify(coordinator, times(rebalanceNum)).configSnapshot(); + verify(coordinator, times(rebalanceNum)).leaderState(any()); + verify(coordinator, times(2 * rebalanceNum)).generationId(); + verify(coordinator, times(rebalanceNum)).memberId(); + verify(coordinator, times(rebalanceNum)).lastCompletedGenerationId(); + } + private WorkerLoad emptyWorkerLoad(String worker) { return new WorkerLoad.Builder(worker).build(); } @@ -1107,20 +1249,32 @@ private static List newConnectors(int start, int end) { } private static List newTasks(int start, int end) { + return newTasks("task", start, end); + } + + private static List newTasks(String connectorName, int start, int end) { return IntStream.range(start, end) - .mapToObj(i -> new ConnectorTaskId("task", i)) + .mapToObj(i -> new ConnectorTaskId(connectorName, i)) .collect(Collectors.toList()); } private static ClusterConfigState clusterConfigState(long offset, int connectorNum, int taskNum) { + return clusterConfigState(offset, 1, connectorNum, taskNum); + } + + private static ClusterConfigState clusterConfigState(long offset, + int connectorStart, + int connectorNum, + int taskNum) { + int connectorNumEnd = connectorStart + connectorNum - 1; return new ClusterConfigState( offset, null, - connectorTaskCounts(1, connectorNum, taskNum), - connectorConfigs(1, connectorNum), - connectorTargetStates(1, connectorNum, TargetState.STARTED), + connectorTaskCounts(connectorStart, connectorNumEnd, taskNum), + connectorConfigs(connectorStart, connectorNumEnd), + connectorTargetStates(connectorStart, connectorNumEnd, TargetState.STARTED), taskConfigs(0, connectorNum, connectorNum * taskNum), Collections.emptySet()); }