private StateUpdater.RemovedTaskResult waitForFuture(final TaskId taskId,
                                                         final CompletableFuture<StateUpdater.RemovedTaskResult> future) {
        final StateUpdater.RemovedTaskResult removedTaskResult;
        try {
            removedTaskResult = future.get();
            if (removedTaskResult == null) {
                throw new IllegalStateException("Task " + taskId + " was not found in the state updater. "
                    + BUG_ERROR_MESSAGE);
            }
            return removedTaskResult;
        } catch (final ExecutionException executionException) {
            log.warn("An exception happened when removing task {} from the state updater. The task was added to the " +
                    "failed task in the state updater: ",
                taskId, executionException);
            return null;
        } catch (final InterruptedException shouldNotHappen) {
            Thread.currentThread().interrupt();
            log.error(INTERRUPTED_ERROR_MESSAGE, shouldNotHappen);
            throw new IllegalStateException(INTERRUPTED_ERROR_MESSAGE, shouldNotHappen);
        }
    }
@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