Copilot commented on code in PR #23556: URL: https://github.com/apache/kafka/pull/23556#discussion_r4123722639
########## group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/assignor/TaskAssignorTestbed.java: ########## @@ -0,0 +1,781 @@ +/* + * 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.assignor; + +import org.apache.kafka.coordinator.group.api.streams.assignor.GroupAssignment; +import org.apache.kafka.coordinator.group.api.streams.assignor.MemberAssignment; +import org.apache.kafka.coordinator.group.api.streams.assignor.TaskAssignor; +import org.apache.kafka.coordinator.group.api.streams.assignor.TopologyDescriber; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.NoSuchElementException; +import java.util.Objects; +import java.util.Optional; +import java.util.Random; +import java.util.Set; +import java.util.TreeMap; +import java.util.TreeSet; + +/** + * Randomized testbed for {@link TaskAssignor} implementations. Each scenario generates a random topology and group + * from a seed, runs the assignor until the assignment is stable, then applies random rebalance events (processes + * joining, leaving, restarting or changing tags, all processes of one tag value leaving at once, members joining + * or leaving, the standby count changing, subtopologies growing, appearing or disappearing) and converges again. + * Every assignment is checked against {@link AssignmentInvariants} and the assignor-specific checks given by the + * caller; the converged result of each rebalance is graded by {@link AssignmentMetrics}, and the summary is printed + * so two versions of an assignor can be compared on the same scenarios. + * <p> + * Runs are deterministic: all scenarios derive from {@link #DEFAULT_BASE_SEED}. Set + * {@code STREAMS_ASSIGNOR_FUZZ_BASE_SEED=<seed>} to run a different set of scenarios, or + * {@code STREAMS_ASSIGNOR_FUZZ_SEED=<seed>} to replay the single scenario named by a failure. From an IDE the + * system properties {@code streams.assignor.fuzz.base.seed} and {@code streams.assignor.fuzz.seed} work as well; + * Gradle does not forward {@code -D} to the test JVM. + */ +final class TaskAssignorTestbed { + + /** + * An assignor-specific check run on every assignment, in addition to the invariants. {@code last} is true for + * the final assignment of a rebalance, the converged one or the last attempt, so expensive checks can run once. + */ + @FunctionalInterface + interface AssignmentCheck { + void check(Scenario scenario, GroupAssignment result, boolean last); + } + + static final long DEFAULT_BASE_SEED = 42L; + static final List<String> TAG_KEYS = List.of("region", "zone", "rack"); + static final String HOST_TAG = "host"; + + private static final String SEED_PROPERTY = "streams.assignor.fuzz.seed"; + private static final String SEED_ENVIRONMENT_VARIABLE = "STREAMS_ASSIGNOR_FUZZ_SEED"; + private static final String BASE_SEED_PROPERTY = "streams.assignor.fuzz.base.seed"; + private static final String BASE_SEED_ENVIRONMENT_VARIABLE = "STREAMS_ASSIGNOR_FUZZ_BASE_SEED"; + private static final int MAX_EVENTS = 8; + private static final int MAX_CONVERGENCE_ITERATIONS = 10; + private static final long RESTORED_OFFSET = 100L; + /** Mean of {@link Profile#LARGE}'s membersPerProcess draw, used to size the group from the task count. */ + private static final int EXPECTED_MEMBERS_PER_LARGE_PROCESS = 9; + + /** + * How large the generated groups are. Each bound is drawn per scenario (or per process) from the given random. + */ + enum Profile { + /** + * Hits the logical corners: fewer processes than copies, tag values exhausted, quota boundaries, load ties. + * Small enough that a failure prints every assignment. + */ + SMALL { + int scenarios() { + return 300; + } + + boolean fullHistory() { + return true; + } + + int processes(final Random random, final int tasks) { + return random.nextInt(8) + 1; + } + + int membersPerProcess(final Random random) { + return random.nextInt(4) + 1; + } + + int subtopologies(final Random random) { + return random.nextInt(6) + 1; + } + + int partitions(final Random random) { + return random.nextInt(8) + 1; + } + + boolean stateful(final Random random) { + return random.nextBoolean(); + } + + int tagKeyCount(final Random random) { + return random.nextInt(TAG_KEYS.size() + 1); + } + + int valuesPerTag(final Random random) { + return random.nextInt(3) + 2; + } + + boolean hostTag(final Random random) { + return false; + } + + int standbyReplicas(final Random random) { + return random.nextInt(3); + } + }, + /** + * Production shape: tens to over a hundred processes with a few large ones, sized so that members hold a + * few active tasks each and the quota leaves room for balancing decisions; one or two tags with about three + * values each so tag dimensions rarely run out, standbys mostly 1. Occasionally adds a host tag whose value + * is unique per process. + */ + LARGE { + int scenarios() { + return 30; + } + + boolean fullHistory() { + return false; + } + + int processes(final Random random, final int tasks) { + final int draw = random.nextInt(10); + final int tasksPerMember = draw < 2 ? 2 : draw < 5 ? 3 : draw < 8 ? 4 : draw < 9 ? 6 : 8; + return Math.max(2, tasks / tasksPerMember / EXPECTED_MEMBERS_PER_LARGE_PROCESS); + } + + int membersPerProcess(final Random random) { + return random.nextInt(10) == 0 ? random.nextInt(33) + 32 : random.nextInt(8) + 1; + } + + int subtopologies(final Random random) { + return random.nextInt(17) + 4; + } + + int partitions(final Random random) { + return random.nextInt(113) + 16; + } + + boolean stateful(final Random random) { + return random.nextInt(10) < 6; + } + + int tagKeyCount(final Random random) { + return random.nextInt(2) + 1; + } + + int valuesPerTag(final Random random) { + final int draw = random.nextInt(10); + return draw < 6 ? 3 : draw < 8 ? 2 : 4; + } + + boolean hostTag(final Random random) { + return random.nextInt(5) == 0; + } + + int standbyReplicas(final Random random) { + final int draw = random.nextInt(10); + return draw < 2 ? 0 : draw < 8 ? 1 : 2; + } + }; + + abstract int scenarios(); + + /** Whether the history records every assignment, or only the metrics of each converged result. */ + abstract boolean fullHistory(); + + /** How many processes to generate for a topology of the given size. */ + abstract int processes(Random random, int tasks); + + abstract int membersPerProcess(Random random); + + abstract int subtopologies(Random random); + + abstract int partitions(Random random); + + abstract boolean stateful(Random random); + + abstract int tagKeyCount(Random random); + + abstract int valuesPerTag(Random random); + + abstract boolean hostTag(Random random); + + abstract int standbyReplicas(Random random); + } + + private final TaskAssignor assignor; + private final List<AssignmentCheck> checks; + + TaskAssignorTestbed(final TaskAssignor assignor, final List<AssignmentCheck> checks) { + this.assignor = assignor; + this.checks = List.copyOf(checks); + } + + /** + * Runs all scenarios of the profile and prints two summaries: the first assignment of each group, computed from + * an empty assignment, and the rebalances after the random events, computed from the previous assignment. + */ + void run(final Profile profile) { + final Long fixedSeed = readSeed(SEED_PROPERTY, SEED_ENVIRONMENT_VARIABLE); + final Long fixedBaseSeed = readSeed(BASE_SEED_PROPERTY, BASE_SEED_ENVIRONMENT_VARIABLE); + final long baseSeed; + if (fixedSeed != null) { + baseSeed = fixedSeed; + } else { + // Offset per profile so the runs do not replay the same scenarios. + final long base = fixedBaseSeed != null ? fixedBaseSeed : DEFAULT_BASE_SEED; + baseSeed = base + profile.ordinal() * 1_000_000L; + } + final int scenarios = fixedSeed != null ? 1 : profile.scenarios(); + final String title = assignor.name() + ", " + profile + ", " + scenarios + " scenarios, base seed " + baseSeed; + final AssignmentMetrics.Summary fromEmpty = new AssignmentMetrics.Summary(title + ", from empty assignment"); + final AssignmentMetrics.Summary fromPrevious = new AssignmentMetrics.Summary(title + ", from previous assignment"); + for (int i = 0; i < scenarios; i++) { + runScenario(baseSeed + i, profile, fromEmpty, fromPrevious); + } + System.out.println(fromEmpty); + System.out.println(fromPrevious); + } + + private static Long readSeed(final String property, final String environmentVariable) { + final Long fromProperty = Long.getLong(property); + if (fromProperty != null) { + return fromProperty; + } + final String fromEnvironment = System.getenv(environmentVariable); + return fromEnvironment == null ? null : Long.valueOf(fromEnvironment); + } + + private void runScenario( + final long seed, + final Profile profile, + final AssignmentMetrics.Summary fromEmpty, + final AssignmentMetrics.Summary fromPrevious + ) { + final Random random = new Random(seed); + final Scenario scenario = Scenario.generate(random, profile); + try { + converge(scenario, fromEmpty, scenario.baseline()); + final int events = random.nextInt(MAX_EVENTS) + 1; + for (int i = 0; i < events; i++) { + final Scenario.Baseline before = scenario.baseline(); + scenario.applyRandomEvent(random); + converge(scenario, fromPrevious, before); + } + } catch (final AssertionError | RuntimeException e) { + throw new AssertionError( + "Fuzz scenario failed. Reproduce with " + SEED_ENVIRONMENT_VARIABLE + "=" + seed + " (or -D" + SEED_PROPERTY + "=" + seed + + ") in the " + profile + " test\n" + scenario.history, + e + ); + } + } + + /** + * Runs the assignor, feeding each result back as the previous assignment, until the result stops changing or + * {@link #MAX_CONVERGENCE_ITERATIONS} is reached. Not reaching a fixed point is recorded, not failed: the + * coordinator runs the assignor once per rebalance, so it only means the next rebalance will move tasks again. + * {@code before} is the state before the event that triggered this rebalance; the last result is graded against it. + */ + private void converge(final Scenario scenario, final AssignmentMetrics.Summary summary, final Scenario.Baseline before) { + // Reported in the first heartbeat after a restart and gone once the members hold an assignment again. + final Map<String, Set<TaskId>> restoredTasks = scenario.restoredTasksByProcess(); + for (int iteration = 1; iteration <= MAX_CONVERGENCE_ITERATIONS; iteration++) { + final GroupAssignment result = assignor.assign(scenario.groupSpec(), scenario.topology); + final boolean stable = result.members().equals(scenario.previousAssignment); + final boolean last = stable || iteration == MAX_CONVERGENCE_ITERATIONS; + try { + AssignmentInvariants.assertValid(scenario, result); + for (final AssignmentCheck check : checks) { + check.check(scenario, result, last); + } + } catch (final AssertionError e) { + scenario.history.append(" iteration ").append(iteration).append(" input: ").append(format(scenario.previousAssignment)).append('\n') + .append(" iteration ").append(iteration).append(" FAILED: ").append(format(result.members())).append('\n'); + throw e; + } + if (scenario.profile.fullHistory()) { + scenario.history.append(" iteration ").append(iteration).append(": ").append(format(result.members())).append('\n'); + } + scenario.feedBack(result); + if (last) { + final AssignmentMetrics.Metrics metrics = AssignmentMetrics.compute(scenario, result, before, restoredTasks); + summary.add(metrics); + if (stable) { + summary.addConvergence(iteration); + scenario.history.append(" converged after ").append(iteration).append(": ").append(metrics).append('\n'); + } else { + summary.addNotConverged(); + scenario.history.append(" not converged within ").append(iteration).append(": ").append(metrics).append('\n'); + } Review Comment: The cutoff does not fail the test, so an assignor that keeps changing its assignment forever still passes both `shouldConverge...` tests. This contradicts their stated convergence guarantee and the existing convergence test pattern (`streams/.../TaskAssignorConvergenceTest.java:622-624`). Please fail after recording the final diagnostic so oscillation regressions are detected. -- 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]
