private void updateInputPartitionsOfStandbyTaskIfTheyChanged(final Task task,
                                                                 final Set<TopicPartition> inputPartitions) {
        /*
        We should only update input partitions of a standby task if the input partitions really changed. Updating the
        input partitions of tasks also updates the mapping from source nodes to input topics in the processor topology
        within the task. The mapping is updated with the topics from the topology metadata. The topology metadata does
        not prefix intermediate internal topics with the application ID. Thus, if a standby task has input partitions
        from an intermediate internal topic the update of the mapping in the processor topology leads to an invalid
        topology exception during recycling of a standby task to an active task when the input queues are created. This
        is because the input topics in the processor topology and the input partitions of the task do not match because
        the former miss the application ID prefix.
        For standby task that have only input partitions from intermediate internal topics this check avoids the invalid
        topology exception. Unfortunately, a subtopology might have input partitions subscribed to with a regex
        additionally intermediate internal topics which might still lead to an invalid topology exception during recycling
        irrespectively of this check here. Thus, there is still a bug to fix here.
         */
        if (!task.inputPartitions().equals(inputPartitions)) {
            task.updateInputPartitions(inputPartitions, topologyMetadata.nodeToSourceTopics(task.id()));
        }
    }
@throws TaskMigratedException if the task producer got fenced (EOS only)
@throws StreamsException fatal error while creating / initializing the task

public for upgrade testing only
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java