private void updateInputPartitions(final Map<TaskId, CompletableFuture<StateUpdater.RemovedTaskResult>> futures,
                                       final Map<TaskId, Set<TopicPartition>> newInputPartitions,
                                       final Map<TaskId, RuntimeException> failedTasks) {
        getNonFailedTasks(futures, failedTasks).forEach(task -> {
            task.updateInputPartitions(
                newInputPartitions.get(task.id()),
                topologyMetadata.nodeToSourceTopics(task.id())
            );
            stateUpdater.add(task);
        });
    }
@Test
    public void shouldNotUpdateExistingStandbyTaskIfStandbyIsReassignedWithSameInputPartitionWithoutStateUpdater() {
        final StandbyTask standbyTask = standbyTask(taskId03, taskId03ChangelogPartitions)
            .inState(State.RUNNING)
            .withInputPartitions(taskId03Partitions).build();
        updateExistingStandbyTaskIfStandbyIsReassignedWithoutStateUpdater(standbyTask, taskId03Partitions);
        verify(standbyTask, never()).updateInputPartitions(eq(taskId03Partitions), any());
    }