Skip to content

KAFKA-9841: Revoke duplicate connectors and tasks when zombie workers return with an outdated assignment #8453

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kkonstantine merged 3 commits into from
Jun 11, 2020
Merged

KAFKA-9841: Revoke duplicate connectors and tasks when zombie workers return with an outdated assignment #8453

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
00bacb7
KAFKA-9841: Connector and Task duplicated when a worker join with old…
LucentWang Apr 9, 2020
76bd2ce
KAFKA-9841: Connector and Task duplicated when a worker join with old…
LucentWang Apr 9, 2020
1621e83
KAFKA-9841: Rephase the comment and fix the unit test
LucentWang Jun 10, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 59 additions & 0 deletions ...ain/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -244,6 +244,10 @@ protected Map<String, ByteBuffer> performTaskAssignment(String leaderId, long ma
Map<String, ConnectorsAndTasks> 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 =
Expand Down Expand Up @@ -359,6 +363,61 @@ private ConnectorsAndTasks computePreviousAssignment(Map<String, ConnectorsAndTa
return previousAssignment;
}

private ConnectorsAndTasks duplicatedAssignments(Map<String, ExtendedWorkerState> memberConfigs) {
Set<String> 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<ConnectorTaskId> 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<String, ConnectorsAndTasks> computeDuplicatedAssignments(Map<String, ExtendedWorkerState> memberConfigs,
Map<String, Collection<String>> connectorAssignments,
Map<String, Collection<ConnectorTaskId>> taskAssignment) {
ConnectorsAndTasks duplicatedAssignments = duplicatedAssignments(memberConfigs);
log.debug("Duplicated assignments: {}", duplicatedAssignments);

Map<String, ConnectorsAndTasks> toRevoke = new HashMap<>();
if (!duplicatedAssignments.connectors().isEmpty()) {
connectorAssignments.entrySet().stream()
.forEach(entry -> {
Set<String> 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<ConnectorTaskId> 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,
Expand Down
162 changes: 158 additions & 4 deletions ...java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
}
Expand All @@ -1107,20 +1249,32 @@ private static List<String> newConnectors(int start, int end) {
}

private static List<ConnectorTaskId> newTasks(int start, int end) {
return newTasks("task", start, end);
}

private static List<ConnectorTaskId> 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());
}
Expand Down