private void initializeRemainingPrevTasksFromTaskOffsetSums() {
        final Set<TaskId> previousActiveTaskIds = previousActiveTasks.taskIds();
        if (previousActiveTaskIds.isEmpty() && !ownedPartitions.isEmpty()) {
            LOG.error("Tried to process tasks in offset sum map before processing tasks from ownedPartitions = {}", ownedPartitions);
            throw new IllegalStateException("Must initialize prevActiveTasks from ownedPartitions before initializing remaining tasks.");
        }
        for (final Map.Entry<TaskId, Long> taskEntry : taskOffsetSums.entrySet()) {
            final TaskId task = taskEntry.getKey();
            if (!previousActiveTaskIds.contains(task)) {
                final long offsetSum = taskEntry.getValue();
                if (offsetSum == Task.LATEST_OFFSET) {
                    addPreviousActiveTask(task);
                } else {
                    addPreviousStandbyTask(task);
                }
            }
        }
    }
@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