private void runRandomizedScenario(final long seed, final String rackAwareStrategy) {
        Harness harness = null;
        try {
            final Random prng = new Random(seed);

            // These are all rand(limit)+1 because we need them to be at least 1 and the upper bound is exclusive
            final int initialClusterSize = prng.nextInt(10) + 1;
            final int numStatelessTasks = prng.nextInt(10) + 1;
            final int numStatefulTasks = prng.nextInt(10) + 1;
            final int maxWarmupReplicas = prng.nextInt(numStatefulTasks) + 1;
            // This one is rand(limit+1) because we _want_ to test zero and the upper bound is exclusive
            final int numStandbyReplicas = prng.nextInt(initialClusterSize + 1);
            final int numNodes = numStatefulTasks + numStatelessTasks;

            final int numberOfEvents = prng.nextInt(10) + 1;

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

            harness = Harness.initializeCluster(
                numStatelessTasks,
                numStatefulTasks,
                initialClusterSize,
                () -> prng.nextInt(10) + 1,
                numNodes
            );
            testForConvergence(harness, configs, 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);
            }

            for (int i = 0; i < numberOfEvents; i++) {
                final int event = prng.nextInt(2);
                switch (event) {
                    case 0:
                        harness.dropRandomClients(prng.nextInt(initialClusterSize), prng);
                        break;
                    case 1:
                        harness.addOrResurrectClientsRandomly(prng, initialClusterSize);
                        break;
                    default:
                        throw new IllegalStateException("Unexpected event: " + event);
                }
                if (!harness.clientStates.isEmpty()) {
                    testForConvergence(harness, configs, 2 * (numStatefulTasks + numStatefulTasks * numStandbyReplicas));
                    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);
                    }
                }
            }
        } catch (final AssertionError t) {
            throw new AssertionError(
                "Assertion failed in randomized test. Reproduce with: `runRandomizedScenario(" + seed + ")`.",
                t
            );
        } catch (final Throwable t) {
            final StringBuilder builder =
                new StringBuilder()
                    .append("Exception in randomized scenario. Reproduce with: `runRandomizedScenario(")
                    .append(seed)
                    .append(")`. ");
            if (harness != null) {
                builder.append(harness.history);
            }
            throw new AssertionError(builder.toString(), t);
        }
    }
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