gabriellefu commented on code in PR #23588:
URL: https://github.com/apache/kafka/pull/23588#discussion_r4404599694
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -132,17 +136,22 @@ private static LocalState initialize(final GroupSpec
groupSpec, final TopologyDe
localState.totalTasksPerMember =
computeTasksPerMember(localState.totalTasks,
localState.totalMembersWithTaskCapacity);
localState.processIdToState =
Utils.newHashMap(localState.totalMembersWithActiveTaskCapacity);
+ localState.processIdToClientTags =
Utils.newHashMap(localState.totalMembersWithActiveTaskCapacity);
localState.activeTaskToPrevMember =
Utils.newHashMap(localState.totalActiveTasks);
+ localState.statefulActiveTaskToProcess =
Utils.newHashMap(localState.totalStatefulActiveTasks);
// Standby-strength candidates per task, gathered in a single pass
over the members and ranked below.
final Map<TaskId, ArrayList<StandbyCandidate>> standbyCandidates = new
HashMap<>();
for (final String memberId : groupSpec.memberIds()) {
final MemberAssignmentState memberAssignmentState =
groupSpec.memberAssignmentState(memberId);
- final String processId =
groupSpec.memberMetadata(memberId).processId();
+ final MemberAssignmentMetadata memberMetadata =
groupSpec.memberMetadata(memberId);
+ final String processId = memberMetadata.processId();
final Member member = new Member(processId, memberId);
localState.processIdToState.computeIfAbsent(processId,
ProcessState::new)
.addMember(memberId);
+ // Client tags belong to the process, so every member of it
reports the same ones.
+ localState.processIdToClientTags.putIfAbsent(processId,
memberMetadata.clientTags());
Review Comment:
thanks for the suggestion, I will update the comment
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
