public SortedSet<TaskId> prevTasksByLag(final String consumer) {
        final SortedSet<TaskId> prevTasksByLag = new TreeSet<>(comparingLong(this::lagFor).thenComparing(TaskId::compareTo));
        for (final TaskId task : prevOwnedStatefulTasksByConsumer(consumer)) {
            if (taskLagTotals.containsKey(task)) {
                prevTasksByLag.add(task);
            } else {
                LOG.debug("Skipping previous task {} since it's not part of the current assignment", task);
            }
        }
        return prevTasksByLag;
    }
@return the previous tasks assigned to this consumer ordered by lag, filtered for any tasks that don't exist in this assignment
streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/ClientState.java