@Override
    public CompletableFuture<RemovedTaskResult> remove(final TaskId taskId) {
        final CompletableFuture<RemovedTaskResult> future = new CompletableFuture<>();
        tasksAndActionsLock.lock();
        try {
            tasksAndActions.add(TaskAndAction.createRemoveTask(taskId, future));
            tasksAndActionsCondition.signalAll();
        } finally {
            tasksAndActionsLock.unlock();
        }
        return future;
    }
@Test
    public void shouldThrowIfRemovingUpdatingStandbyTaskFailsWithStreamsException() throws Exception {
        final StandbyTask task = standbyTask(TASK_0_0, mkSet(TOPIC_PARTITION_A_0)).inState(State.RUNNING).build();
        final StreamsException streamsException = new StreamsException("Something happened", task.id());
        setupShouldThrowIfRemovingUpdatingStatefulTaskFailsWithException(task, streamsException);

        final CompletableFuture<StateUpdater.RemovedTaskResult> future = stateUpdater.remove(task.id());

        verifyRemovingUpdatingStatefulTaskFails(future, task, streamsException, true);
    }