private Map<TaskId, Set<TopicPartition>> buildStandbyTaskMap(final String consumer,
                                                                 final Iterable<TaskId> standbyTasks,
                                                                 final Iterable<TaskId> revokedTasks,
                                                                 final Set<TaskId> allStatefulTasks,
                                                                 final Map<TaskId, Set<TopicPartition>> partitionsForTask,
                                                                 final ClientState clientState) {
        final Map<TaskId, Set<TopicPartition>> standbyTaskMap = new HashMap<>();

        for (final TaskId task : standbyTasks) {
            clientState.assignStandbyToConsumer(task, consumer);
            standbyTaskMap.put(task, partitionsForTask.get(task));
        }

        for (final TaskId task : revokedTasks) {
            // If this task is stateful and already owned by the consumer, but can't (yet) be assigned as an active
            // task during this rebalance as it must be revoked from another consumer first, place a temporary
            // standby task here until it can receive the active task to avoid closing the state store (and losing
            // all of the accumulated state in the case of in-memory stores)
            if (clientState.previouslyOwnedStandby(task) && allStatefulTasks.contains(task)) {
                log.info("Adding removed stateful active task {} as a standby for {} until it is revoked and can "
                             + "be transitioned to active in a followup rebalance", task, consumer);

                // This has no effect on the assignment, as we'll never consult the ClientState again, but
                // it does perform a useful assertion that the it's legal to assign this task as a standby to this instance
                clientState.assignStandbyToConsumer(task, consumer);
                clientState.assignStandby(task);

                standbyTaskMap.put(task, partitionsForTask.get(task));
            }
        }
        return standbyTaskMap;
    }
@return map from task id to its assigned partitions for all standby tasks
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java