private static void verifyBalancedAssignment(final Harness harness, final int skewThreshold) {
        final Set<TaskId> allStatefulTasks = harness.statefulTaskEndOffsetSums.keySet();
        final Map<ProcessId, ClientState> clientStates = harness.clientStates;
        final StringBuilder failureContext = harness.history;

        assertBalancedActiveAssignment(clientStates, failureContext);
        assertBalancedStatefulAssignment(allStatefulTasks, clientStates, failureContext);
        final AssignmentTestUtils.TaskSkewReport taskSkewReport = AssignmentTestUtils.analyzeTaskAssignmentBalance(harness.clientStates, skewThreshold);
        if (taskSkewReport.totalSkewedTasks() > 0) {
            fail(
                new StringBuilder().append("Expected a balanced task assignment, but was: ")
                                   .append(taskSkewReport)
                                   .append('\n')
                                   .append(failureContext)
                                   .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