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 b81c2473653c9805902649256e5ce422eed41a07 Author: Benoit TELLIER <[email protected]> AuthorDate: Sun Aug 16 16:38:40 2026 +0700 [FIX] Allow canceling tasks with truncated history --- .../james/task/eventsourcing/CommandHandlers.scala | 10 ++++-- .../EventSourcingTaskManagerTest.java | 39 +++++++++++++++++++++- 2 files changed, 46 insertions(+), 3 deletions(-) diff --git a/server/task/task-memory/src/main/scala/org/apache/james/task/eventsourcing/CommandHandlers.scala b/server/task/task-memory/src/main/scala/org/apache/james/task/eventsourcing/CommandHandlers.scala index e1c4ab8309..549c618b76 100644 --- a/server/task/task-memory/src/main/scala/org/apache/james/task/eventsourcing/CommandHandlers.scala +++ b/server/task/task-memory/src/main/scala/org/apache/james/task/eventsourcing/CommandHandlers.scala @@ -59,7 +59,13 @@ class RequestCancelCommandHandler(private val loadHistory: TaskAggregateId => SM override def handledClass: Class[RequestCancel] = classOf[RequestCancel] override def handle(command: RequestCancel): Publisher[List[EventWithState]] = { - loadAggregate(loadHistory, command.id).map(_.requestCancel(hostname).map(EventWithState.noState).asJava) + val aggregateId = TaskAggregateId(command.id) + loadHistory(aggregateId).map(history => + if (history.getEvents.isEmpty) { + Seq.empty[EventWithState].asJava + } else { + TaskAggregate.fromHistory(aggregateId, history).requestCancel(hostname).map(EventWithState.noState).asJava + }) } } @@ -93,4 +99,4 @@ class UpdateCommandHandler(private val loadHistory: TaskAggregateId => SMono[His override def handle(command: UpdateAdditionalInformation): Publisher[List[EventWithState]] = { loadAggregate(loadHistory, command.id).map(_.update(command.additionalInformation).map(EventWithState.noState).asJava) } -} \ No newline at end of file +} 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 fb6f30ff1d..f54f61d042 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 @@ -20,6 +20,11 @@ package org.apache.james.task.eventsourcing; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; + +import java.time.ZonedDateTime; +import java.time.temporal.ChronoUnit; +import java.util.Optional; import org.apache.james.eventsourcing.eventstore.EventStore; import org.apache.james.eventsourcing.eventstore.memory.InMemoryEventStore; @@ -29,10 +34,12 @@ import org.apache.james.task.MemoryReferenceTask; import org.apache.james.task.MemoryWorkQueue; import org.apache.james.task.SerialTaskManagerWorker; import org.apache.james.task.Task; +import org.apache.james.task.TaskExecutionDetails; import org.apache.james.task.TaskId; import org.apache.james.task.TaskManager; import org.apache.james.task.TaskManagerContract; import org.apache.james.task.TaskManagerWorker; +import org.apache.james.task.TaskType; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -45,11 +52,12 @@ class EventSourcingTaskManagerTest implements TaskManagerContract { private static final Hostname HOSTNAME = new Hostname("foo"); private EventSourcingTaskManager taskManager; private EventStore eventStore; + private TaskExecutionDetailsProjection executionDetailsProjection; @BeforeEach void setUp() { eventStore = new InMemoryEventStore(); - TaskExecutionDetailsProjection executionDetailsProjection = new MemoryTaskExecutionDetailsProjection(); + executionDetailsProjection = new MemoryTaskExecutionDetailsProjection(); WorkQueueSupplier workQueueSupplier = eventSourcingSystem -> { WorkerStatusListener listener = new WorkerStatusListener(eventSourcingSystem); TaskManagerWorker worker = new SerialTaskManagerWorker(listener, UPDATE_INFORMATION_POLLING_INTERVAL); @@ -105,4 +113,33 @@ class EventSourcingTaskManagerTest implements TaskManagerContract { .extracting("hostname") .containsOnly(HOSTNAME)); } + + @Test + void cancelShouldNotFailWhenExecutionDetailsHaveNoEvents() { + TaskId taskId = TaskId.generateTaskId(); + executionDetailsProjection.update(unfinishedDetailsWithoutEvents(taskId)); + + assertThatCode(() -> taskManager.cancel(taskId)) + .doesNotThrowAnyException(); + } + + /** + * An entry of the execution details projection whose events are missing from the event store: such an + * inconsistency is caused by a data loss on the event store, and used to make the task impossible to + * interact with at all. + */ + private TaskExecutionDetails unfinishedDetailsWithoutEvents(TaskId taskId) { + return new TaskExecutionDetails(taskId, + TaskType.of("type"), + TaskManager.Status.IN_PROGRESS, + ZonedDateTime.now().minus(20, ChronoUnit.DAYS), + HOSTNAME, + Optional::empty, + Optional.empty(), + Optional.empty(), + Optional.empty(), + Optional.empty(), + Optional.empty(), + Optional.empty()); + } } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
