@ParameterizedTest
    @ValueSource(strings = {
        StreamsConfig.RACK_AWARE_ASSIGNMENT_STRATEGY_NONE,
        StreamsConfig.RACK_AWARE_ASSIGNMENT_STRATEGY_MIN_TRAFFIC,
        StreamsConfig.RACK_AWARE_ASSIGNMENT_STRATEGY_BALANCE_SUBTOPOLOGY
    })
    public void assignmentShouldConvergeAfterAddingNode(final String rackAwareStrategy) {
        setUp(rackAwareStrategy);
        final int numStatelessTasks = 7;
        final int numStatefulTasks = 11;
        final int maxWarmupReplicas = 2;
        final int numStandbyReplicas = 0;
        final int numNodes = 10;

        final AssignmentConfigs configs = new AssignmentConfigs(100L,
                                                                maxWarmupReplicas,
                                                                numStandbyReplicas,
                                                                60_000L,
                                                                EMPTY_RACK_AWARE_ASSIGNMENT_TAGS,
                                                                null,
                                                                null,
                                                                rackAwareStrategy);

        final Harness harness = Harness.initializeCluster(numStatelessTasks, numStatefulTasks, 1, () -> 5, numNodes);
        testForConvergence(harness, configs, 1);
        harness.addClient();
        // we expect convergence to involve moving each task at most once, and we can move "maxWarmupReplicas" number
        // of tasks at once, hence the iteration limit
        testForConvergence(harness, configs, numStatefulTasks / maxWarmupReplicas + 1);
        verifyValidAssignment(numStandbyReplicas, harness);

        // min-cost rack aware assignor doesn't balance subtopolgy
        if (!rackAwareStrategy.equals(StreamsConfig.RACK_AWARE_ASSIGNMENT_STRATEGY_MIN_TRAFFIC)) {
            verifyBalancedAssignment(harness, skewThreshold);
        }
    }
Flip the cluster states from "assigned" to "subscribed" so they can be used for another round of assignments.
streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/TaskAssignorConvergenceTest.java