mjsax commented on code in PR #23342: URL: https://github.com/apache/kafka/pull/23342#discussion_r3921598533
########## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/LagDrivenAssignmentRefiner.java: ########## @@ -0,0 +1,256 @@ +/* + * 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.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.Set; +import java.util.SortedMap; +import java.util.TreeMap; + +/** + * An {@link AssignmentRefiner} that holds a stateful task back with its current owner until the member the task is + * moving to has restored it as a warm-up task. Intended for testing and as a starting point for a real refiner; it is + * deliberately simple and is <b>not</b> the derivation the broker will ship. + * <p> + * For every stateful task whose target owner is not its current owner, it withholds the task from the target owner -- + * leaving it active where it already runs -- and hands the target owner a warm-up task instead, so it can restore the + * state in the background. Once that member reports a lag within {@code acceptableRecoveryLag} for the task, the + * withholding stops and the target assignment flows through, at which point the reconciler revokes the task from the + * old owner and promotes the warm-up in place. {@code numWarmupReplicas} caps how many warm-up tasks it hands out at a + * time; a migration that does not get a slot is still held back, it just runs without anybody warming it up. + * <p> + * Three cases are deliberately let through without warming up, because there is no state to restore first: + * <ul> + * <li>a stateless task, which has no state,</li> + * <li>a task moving between two members of the same process, whose state directory is already local, and</li> + * <li>a task whose old owner was already told to revoke it, which means the hand-over is under way and reversing it + * would strand the task with nobody running it.</li> + * </ul> + * <p> + * What it does <em>not</em> do, and a real refiner has to: rank migrations by how loaded their source and destination + * are (this one orders them by task ID, so the warm-up budget is handed out in an arbitrary but stable order), + * carry warm-up slots over from the previous step rather than re-deriving them, convert an existing standby into a + * warm-up instead of spending a slot on a new one, and defer standbys that cannot be placed yet instead of dropping + * them. It also assumes the assignor is sticky enough not to keep re-targeting a task while it is being warmed up. + */ +public class LagDrivenAssignmentRefiner implements AssignmentRefiner { Review Comment: This is actually a test-only class, but I added it to "main" module to make it easy to load in an integration test. Module `streams-integration` test only depends on the main module of `group-coordinator`, not the test module. Happy to move elsewhere if preferred; I just followed the path of least resistance :) -- Actually similar to the existing `MockAssignor.java` in the same module. -- 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]
