@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 droppingNodesShouldConverge(final String rackAwareStrategy) {
        setUp(rackAwareStrategy);
        final int numStatelessTasks = 11;
        final int numStatefulTasks = 13;
        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, 7, () -> 5, numNodes);
        testForConvergence(harness, configs, 1);
        harness.dropClient();
        // This time, we allow one extra iteration because the
        // first stateful task needs to get shuffled back to the first node
        testForConvergence(harness, configs, numStatefulTasks / maxWarmupReplicas + 2);

        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