private static void testForConvergence(final Harness harness,
                                           final AssignmentConfigs configs,
                                           final int iterationLimit) {
        final Set<TaskId> allTasks = new TreeSet<>();
        allTasks.addAll(harness.statelessTasks);
        allTasks.addAll(harness.statefulTaskEndOffsetSums.keySet());

        harness.recordConfig(configs);

        boolean rebalancePending = true;
        int iteration = 0;
        final RackAwareTaskAssignor rackAwareTaskAssignor = new RackAwareTaskAssignor(
            harness.fullMetadata,
            harness.partitionsForTask,
            harness.changelogPartitionsForTask,
            harness.tasksForTopicGroup,
            harness.racksForProcessConsumer,
            harness.internalTopicManager,
            configs,
            TIME
        );
        while (rebalancePending && iteration < iterationLimit) {
            iteration++;
            harness.prepareForNextRebalance();
            harness.recordBefore(iteration);
            rebalancePending = new HighAvailabilityTaskAssignor().assign(
                harness.clientStates,
                allTasks,
                harness.statefulTaskEndOffsetSums.keySet(),
                rackAwareTaskAssignor,
                configs
            );
            harness.recordAfter(iteration, rebalancePending);
        }

        if (rebalancePending) {
            final StringBuilder message =
                new StringBuilder().append("Rebalances have not converged after iteration cutoff: ")
                                   .append(iterationLimit)
                                   .append(harness.history);
            fail(message.toString());
        }
    }
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