lucasbru commented on code in PR #23325: URL: https://github.com/apache/kafka/pull/23325#discussion_r3934104136
########## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java: ########## @@ -0,0 +1,506 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.coordinator.group.streams; + +import org.apache.kafka.coordinator.group.streams.assignor.TaskId; +import org.apache.kafka.coordinator.group.streams.topics.ConfiguredSubtopology; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.SortedMap; +import java.util.TreeMap; +import java.util.function.Consumer; + +/** + * The {@link AssignmentRefiner} being built out to replace {@link NoOpAssignmentRefiner} as the broker's default + * once the derivation is complete. + * + * <p>{@link #refine} is still a stub -- it returns the target assignment unchanged, exactly like + * {@link NoOpAssignmentRefiner} -- while the derivation is built out incrementally across several changes. The + * methods below are its building blocks: indexing the current assignment, and deciding which migrations can + * complete immediately versus which have to stage behind a warm-up task. None of them are called from + * {@link #refine} yet. + */ +public class AssignmentRefinerImpl implements AssignmentRefiner { + + @Override + public Map<String, TasksTuple> refine( + Map<String, StreamsGroupMember> members, + Map<String, TasksTuple> targetAssignment, + Map<String, MemberTaskOffsets> taskOffsets, + SortedMap<String, ConfiguredSubtopology> subtopologies, + int numWarmupReplicas, + long acceptableRecoveryLag + ) { + return targetAssignment; + } + + /** + * Indexes the members' current assignment by task, so that the case analysis can look up what a task's situation is + * without scanning the group again for every task. This is a single pass over the members' task entries. + * + * <p>Only stateful tasks are indexed. A stateless task has no state to restore, so it is never staged and never + * consulted here; it simply flows through from the target assignment. + * + * @param members + * All members of the group. + * @param taskOffsets + * The latest changelog offsets/end-offsets reported by the members. + * @param subtopologies + * The resolved subtopologies, which tell whether a subtopology is stateful. + * @param acceptableRecoveryLag + * The lag at or below which a replica counts as caught up. + * + * @return The current assignment, indexed by task. + */ + static CurrentAssignmentIndex indexCurrentAssignment( + final Map<String, StreamsGroupMember> members, + final Map<String, MemberTaskOffsets> taskOffsets, + final SortedMap<String, ConfiguredSubtopology> subtopologies, + final long acceptableRecoveryLag + ) { + final Map<TaskId, String> activeOwner = new HashMap<>(); + final Map<TaskId, Set<String>> processesRevoking = new HashMap<>(); + final Map<TaskId, List<TaskCopy>> taskCopies = new HashMap<>(); + + for (final StreamsGroupMember member : members.values()) { + final MemberTaskOffsets offsets = taskOffsets.getOrDefault(member.memberId(), MemberTaskOffsets.EMPTY); + + forEachStatefulActiveTask( + member.assignedTasks().activeTasksWithEpochs(), + subtopologies, + task -> activeOwner.put(task, member.memberId()) + ); + + // The member has been told to give this task up and has stopped running it, so the member is + // deliberately not recorded as the task's active owner. Recording the member as the owner would make the + // case analysis try to keep the task there, undoing a hand-over that is already under way. + // + // The task does still occupy the member's process until the revocation completes, and that is what stops + // a standby of the same task being placed there. The block applies to the whole process, not just this + // one member, so the process is what gets recorded. + forEachStatefulActiveTask( + member.tasksPendingRevocation().activeTasksWithEpochs(), Review Comment: Why do we not care about standby tasks and warmup tasks pending revocation? -- 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]
