class SerialTaskManagerWorker implements TaskManagerWorker {

    @Override
    public Mono<Task.Result> executeTask(TaskWithId taskWithId) {
        if (!cancelledTasks.remove(taskWithId.getId())) {
            Mono<Task.Result> taskMono = Mono.fromCallable(() -> runWithMdc(taskWithId, listener)).subscribeOn(taskExecutor);
            CompletableFuture<Task.Result> future = taskMono.toFuture();
            runningTask.set(Tuples.of(taskWithId.getId(), future));

            return Mono.using(
                () -> pollAdditionalInformation(taskWithId).subscribe(),
                ignored -> Mono.fromFuture(future)
                    .onErrorResume(exception -> Mono.from(handleExecutionError(taskWithId, listener, exception))
                            .thenReturn(Task.Result.PARTIAL)),
                Disposable::dispose);
        } else {
            return Mono.from(listener.cancelled(taskWithId.getId(), taskWithId.getTask().details()))
                .then(Mono.empty());
        }
    }

    public  SerialTaskManagerWorker(Listener listener, Duration pollingInterval);

}