This is an automated email from the ASF dual-hosted git repository. quantranhong1999 pushed a commit to branch 3.9.x in repository https://gitbox.apache.org/repos/asf/james-project.git
commit d3f6ef0b0869bb3106ba11d87e5635fc044f8856 Author: Benoit TELLIER <[email protected]> AuthorDate: Sun Aug 16 23:38:17 2026 +0700 [FIX] Toletate failures for Task additional information --- .../distributed/RabbitMQWorkQueue.java | 3 ++ .../apache/james/task/SerialTaskManagerWorker.java | 20 +++++++- .../task/eventsourcing/WorkerStatusListener.scala | 34 ++++++++++--- .../EventSourcingTaskManagerTest.java | 59 ++++++++++++++++++++++ 4 files changed, 108 insertions(+), 8 deletions(-) diff --git a/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueue.java b/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueue.java index 48179512a4..94500011bf 100644 --- a/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueue.java +++ b/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueue.java @@ -164,6 +164,9 @@ public class RabbitMQWorkQueue implements WorkQueue { .map(taskIdValue -> TaskId.fromString(taskIdValue.toString())) .flatMap(taskId -> Mono.fromCallable(() -> new String(delivery.getBody(), StandardCharsets.UTF_8)) .flatMap(bodyValue -> deserialize(bodyValue, taskId)) + // Deserialization failures yield no task: the delivery would otherwise be neither acked nor + // nacked, and thus be redelivered over and over. + .switchIfEmpty(Mono.fromRunnable(() -> delivery.nack(!REQUEUE))) .doOnNext(task -> delivery.ack()) .flatMap(task -> executeOnWorker(taskId, task)) .doOnSuccess(result -> LOGGER.info("Executed task {} yield {}", taskId, result))) diff --git a/server/task/task-memory/src/main/java/org/apache/james/task/SerialTaskManagerWorker.java b/server/task/task-memory/src/main/java/org/apache/james/task/SerialTaskManagerWorker.java index 6144bd7e88..fee70fcfa2 100644 --- a/server/task/task-memory/src/main/java/org/apache/james/task/SerialTaskManagerWorker.java +++ b/server/task/task-memory/src/main/java/org/apache/james/task/SerialTaskManagerWorker.java @@ -81,7 +81,9 @@ public class SerialTaskManagerWorker implements TaskManagerWorker { runningTasks.put(taskWithId.getId(), future); Mono<Task.Result> pollingMono = Mono.using( - () -> pollAdditionalInformation(taskWithId).subscribe(), + () -> pollAdditionalInformation(taskWithId).subscribe( + any -> { }, + e -> LOGGER.error("Stopped polling additional information updates of task {}", taskWithId.getId().asString(), e)), ignored -> Mono.fromFuture(future) .onErrorResume(exception -> Mono.from(handleExecutionError(taskWithId, listener, exception)) .thenReturn(Task.Result.PARTIAL)), @@ -119,9 +121,23 @@ public class SerialTaskManagerWorker implements TaskManagerWorker { } } + /** + * Polls the progress of a task until it terminates. + * + * Computing the additional information of a task can fail - typically when it is backed by the very data + * the task is struggling with. Such a failure needs to be swallowed: it would otherwise terminate this + * Flux, silently depriving the task of any progress reporting for the rest of its execution. + * + * Note that the delay needs to be applied upon subscription rather than upon emission: an errored, hence + * empty, poll would otherwise complete right away and have `repeat` resubscribe in a tight loop. + */ private Flux<TaskExecutionDetails.AdditionalInformation> pollAdditionalInformation(TaskWithId taskWithId) { return Mono.from(taskWithId.getTask().detailsReactive()) - .delayElement(pollingInterval, Schedulers.parallel()) + .onErrorResume(e -> { + LOGGER.error("Error while computing additional information updates of task {}", taskWithId.getId().asString(), e); + return Mono.empty(); + }) + .delaySubscription(pollingInterval, Schedulers.parallel()) .repeat() .handle(publishIfPresent()) .flatMap(information -> Mono.from(listener.updated(taskWithId.getId(), Mono.just(information))) diff --git a/server/task/task-memory/src/main/scala/org/apache/james/task/eventsourcing/WorkerStatusListener.scala b/server/task/task-memory/src/main/scala/org/apache/james/task/eventsourcing/WorkerStatusListener.scala index 642e6fa1a4..b4e2bc24a4 100644 --- a/server/task/task-memory/src/main/scala/org/apache/james/task/eventsourcing/WorkerStatusListener.scala +++ b/server/task/task-memory/src/main/scala/org/apache/james/task/eventsourcing/WorkerStatusListener.scala @@ -25,30 +25,52 @@ import org.apache.james.task.Task.Result import org.apache.james.task.eventsourcing.TaskCommand._ import org.apache.james.task.{TaskExecutionDetails, TaskId, TaskManagerWorker} import org.reactivestreams.Publisher +import org.slf4j.{Logger, LoggerFactory} import reactor.core.scala.publisher.SMono import java.util.Optional import scala.jdk.OptionConverters._ +object WorkerStatusListener { + private val LOGGER: Logger = LoggerFactory.getLogger(classOf[WorkerStatusListener]) +} + case class WorkerStatusListener(eventSourcingSystem: EventSourcingSystem) extends TaskManagerWorker.Listener { + import WorkerStatusListener.LOGGER + override def started(taskId: TaskId): Publisher[Void] = SMono(eventSourcingSystem.dispatch(Start(taskId))).`then`() override def completed(taskId: TaskId, result: Result, additionalInformationPublisher: Publisher[Optional[TaskExecutionDetails.AdditionalInformation]]): Publisher[Void] = - SMono.fromPublisher(additionalInformationPublisher) - .flatMap(additionalInformation => SMono(eventSourcingSystem.dispatch(Complete(taskId, result, additionalInformation.toScala)))) + additionalInformation(taskId, additionalInformationPublisher) + .flatMap(additionalInformation => SMono(eventSourcingSystem.dispatch(Complete(taskId, result, additionalInformation)))) .`then`() override def failed(taskId: TaskId, additionalInformationPublisher: Publisher[Optional[TaskExecutionDetails.AdditionalInformation]], errorMessage: Optional[String], t: Optional[Throwable]): Publisher[Void] = - SMono.fromPublisher(additionalInformationPublisher) - .flatMap(additionalInformation => SMono(eventSourcingSystem.dispatch(Fail(taskId, additionalInformation.toScala, errorMessage.toScala, t.toScala.map(t => Throwables.getStackTraceAsString(t)))))) + additionalInformation(taskId, additionalInformationPublisher) + .flatMap(additionalInformation => SMono(eventSourcingSystem.dispatch(Fail(taskId, additionalInformation, errorMessage.toScala, t.toScala.map(t => Throwables.getStackTraceAsString(t)))))) .`then`() override def cancelled(taskId: TaskId, additionalInformationPublisher: Publisher[Optional[TaskExecutionDetails.AdditionalInformation]]): Publisher[Void] = - SMono.fromPublisher(additionalInformationPublisher) - .flatMap(additionalInformation => SMono(eventSourcingSystem.dispatch(Cancel(taskId, additionalInformation.toScala)))) + additionalInformation(taskId, additionalInformationPublisher) + .flatMap(additionalInformation => SMono(eventSourcingSystem.dispatch(Cancel(taskId, additionalInformation)))) .`then`() + /** + * A task needs to be able to reach a terminal state even when its progress reporting is broken: some tasks + * compute their additional information by querying the very data they are working on, which can well be the + * failure being recorded. Letting such an error propagate would skip the dispatch altogether, leaving the + * task in progress forever - neither completable, nor failable, nor cancellable. + */ + private def additionalInformation(taskId: TaskId, additionalInformationPublisher: Publisher[Optional[TaskExecutionDetails.AdditionalInformation]]): SMono[Option[TaskExecutionDetails.AdditionalInformation]] = + SMono.fromPublisher(additionalInformationPublisher) + .map(_.toScala) + .onErrorResume(e => { + LOGGER.warn("Could not retrieve additional information of task {}. Recording its outcome without it.", taskId.asString(), e) + SMono.just(None) + }) + .defaultIfEmpty(None) + override def updated(taskId: TaskId, additionalInformationPublisher: Publisher[TaskExecutionDetails.AdditionalInformation]): Publisher[Void] = SMono.fromPublisher(additionalInformationPublisher) .flatMap(additionalInformation => SMono(eventSourcingSystem.dispatch(UpdateAdditionalInformation(taskId, additionalInformation)))) diff --git a/server/task/task-memory/src/test/java/org/apache/james/task/eventsourcing/EventSourcingTaskManagerTest.java b/server/task/task-memory/src/test/java/org/apache/james/task/eventsourcing/EventSourcingTaskManagerTest.java index f54f61d042..8956e93b22 100644 --- a/server/task/task-memory/src/test/java/org/apache/james/task/eventsourcing/EventSourcingTaskManagerTest.java +++ b/server/task/task-memory/src/test/java/org/apache/james/task/eventsourcing/EventSourcingTaskManagerTest.java @@ -25,6 +25,7 @@ import static org.assertj.core.api.Assertions.assertThatCode; import java.time.ZonedDateTime; import java.time.temporal.ChronoUnit; import java.util.Optional; +import java.util.concurrent.CountDownLatch; import org.apache.james.eventsourcing.eventstore.EventStore; import org.apache.james.eventsourcing.eventstore.memory.InMemoryEventStore; @@ -44,6 +45,8 @@ import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.api.function.ThrowingSupplier; +import org.reactivestreams.Publisher; import reactor.core.publisher.Mono; @@ -114,6 +117,62 @@ class EventSourcingTaskManagerTest implements TaskManagerContract { .containsOnly(HOSTNAME)); } + @Test + void taskShouldCompleteWhenAdditionalInformationCanNotBeComputed() { + TaskId taskId = taskManager.submit(new BrokenDetailsTask(() -> Task.Result.COMPLETED)); + + awaitUntilTaskHasStatus(taskId, TaskManager.Status.COMPLETED, taskManager); + } + + @Test + void taskShouldFailWhenAdditionalInformationCanNotBeComputed() { + TaskId taskId = taskManager.submit(new BrokenDetailsTask(() -> Task.Result.PARTIAL)); + + awaitUntilTaskHasStatus(taskId, TaskManager.Status.FAILED, taskManager); + } + + @Test + void taskShouldBeCancelledWhenAdditionalInformationCanNotBeComputed(CountDownLatch countDownLatch) { + TaskId taskId = taskManager.submit(new BrokenDetailsTask(() -> { + countDownLatch.await(); + return Task.Result.COMPLETED; + })); + + awaitUntilTaskHasStatus(taskId, TaskManager.Status.IN_PROGRESS, taskManager); + taskManager.cancel(taskId); + countDownLatch.countDown(); + + awaitUntilTaskHasStatus(taskId, TaskManager.Status.CANCELLED, taskManager); + } + + /** + * A task computing its additional information out of the very data it is working on: its progress reporting + * breaks down exactly when things go wrong. Recording the outcome of such a task needs to keep working, as + * it would otherwise be left in progress forever. + */ + private static class BrokenDetailsTask implements Task { + private final ThrowingSupplier<Result> task; + + BrokenDetailsTask(ThrowingSupplier<Result> task) { + this.task = task; + } + + @Override + public Result run() throws InterruptedException { + return new MemoryReferenceTask(task).run(); + } + + @Override + public TaskType type() { + return TaskType.of("broken-details"); + } + + @Override + public Publisher<Optional<TaskExecutionDetails.AdditionalInformation>> detailsReactive() { + return Mono.error(new RuntimeException("Additional information can not be computed")); + } + } + @Test void cancelShouldNotFailWhenExecutionDetailsHaveNoEvents() { TaskId taskId = TaskId.generateTaskId(); --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
