private Map<TaskId, Set<TopicPartition>> pendingTasksToCreate(final Map<TaskId, Set<TopicPartition>> tasksToCreate) {
        final Map<TaskId, Set<TopicPartition>> pendingTasks = new HashMap<>();
        final Iterator<Map.Entry<TaskId, Set<TopicPartition>>> iter = tasksToCreate.entrySet().iterator();
        while (iter.hasNext()) {
            final Map.Entry<TaskId, Set<TopicPartition>> entry = iter.next();
            final TaskId taskId = entry.getKey();
            final boolean taskIsOwned = tasks.allTaskIds().contains(taskId)
                || (stateUpdater != null && stateUpdater.getTasks().stream().anyMatch(task -> task.id() == taskId));
            if (taskId.topologyName() != null && !taskIsOwned && !topologyMetadata.namedTopologiesView().contains(taskId.topologyName())) {
                log.info("Cannot create the assigned task {} since it's topology name cannot be recognized, will put it " +
                        "aside as pending for now and create later when topology metadata gets refreshed", taskId);
                pendingTasks.put(taskId, entry.getValue());
                iter.remove();
            }
        }
        return pendingTasks;
    }
@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