private void formatClientStates(final boolean printUnassigned) {
            appendClientStates(history, clientStates);
            if (printUnassigned) {
                final Set<TaskId> unassignedTasks = new TreeSet<>();
                unassignedTasks.addAll(statefulTaskEndOffsetSums.keySet());
                unassignedTasks.addAll(statelessTasks);
                for (final Map.Entry<ProcessId, ClientState> entry : clientStates.entrySet()) {
                    unassignedTasks.removeAll(entry.getValue().assignedTasks());
                }
                history.append("Unassigned Tasks: ").append(unassignedTasks).append('\n');
            }
        }
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