Package io.axoniq.framework.workflow.runtime.execution

Class WorkflowEngine

java.lang.Object
io.axoniq.framework.workflow.runtime.execution.WorkflowEngine
All Implemented Interfaces:
WorkflowEngineCheckpointingSupport.CheckpointLatchCoordinator, MessageHandler, EventHandler
@Internal public class WorkflowEngine extends Object implements EventHandler, WorkflowEngineCheckpointingSupport.CheckpointLatchCoordinator
Main workflow component responsible for managing and executing workflows.
Since:
5.4.0
Author:
Allard Buijze, Simon Zambrovski, Stefan Dragisic, Steven van Beelen

  • Constructor Details

  • Method Details

    • setCheckpointingSupport

      @Internal public void setCheckpointingSupport(WorkflowEngineCheckpointingSupport checkpointingSupport)
      Sets the checkpointing support required by this engine.

      The component is set outside the WorkflowEngine(WorkflowConfigurationRegistry, WorkflowExecutionRepository, WorkflowCancellationService, WorkflowStore, UnitOfWorkFactory), because they require this WorkflowEngine itself to function. Hence, a cyclic dependency would exist upon start-up if completion otherwise.

      Parameters:
      checkpointingSupport - provides checkpointing support to this WorkflowEngine

    • handle

      public MessageStream.Empty<Message> handle(EventMessage event, ProcessingContext context)
      Description copied from interface: EventHandler
      Handles the given event within the given context.

      The result of handling is an empty stream.

      Specified by:
      handle in interface EventHandler
      Parameters:
      event - The event to handle.
      context - The context to the given event is handled in.
      Returns:
      An empty stream containing nothing.

    • hasUnsafeCheckpointWork

      public boolean hasUnsafeCheckpointWork(Segment segment)
      Description copied from interface: WorkflowEngineCheckpointingSupport.CheckpointLatchCoordinator
      Returns whether any WorkflowExecutions owned by the given segment still makes checkpoint advancement unsafe.
      Specified by:
      hasUnsafeCheckpointWork in interface WorkflowEngineCheckpointingSupport.CheckpointLatchCoordinator
      Parameters:
      segment - the segment whose checkpoint is being advanced
      Returns:
      true when checkpoint advancement must wait, false otherwise

    • holdsCheckpoint

      public boolean holdsCheckpoint(Segment segment)
      Specified by:
      holdsCheckpoint in interface WorkflowEngineCheckpointingSupport.CheckpointLatchCoordinator

    • addCheckpointLatch

      public void addCheckpointLatch(Segment segment, Runnable latch)
      Adds a checkpoint latch across the current set of WorkflowExecutions owned by the given segment that are still performing tasks.

      The given latch is attached to all WorkflowExecutions that still have tasks to perform. Or in other terms, executions that are "unsafe" to checkpoint on. When no execution is unsafe at the moment the latch is attached, the latch runs straight away so safety is re-checked, covering a workflow that appended work concurrently.

      Specified by:
      addCheckpointLatch in interface WorkflowEngineCheckpointingSupport.CheckpointLatchCoordinator
      Parameters:
      segment - the segment whose checkpoint is being advanced
      latch - the latch to invoke after all unsafe WorkflowExecutions have reached it

    • workflowExecutions

      public Set<WorkflowExecution> workflowExecutions()
      Retrieve all workflow executions.
      Returns:
      set of currently running workflow executions.

    • start

      public CompletableFuture<Void> start(@Nullable TrackingToken latestToken)
      Records the stream position that restored segments must reach before their workflow bodies start.
      Parameters:
      latestToken - latest token observed while starting the processor
      Returns:
      a completed future after startup validation

    • restoreWorkflowsFor

      public CompletableFuture<Void> restoreWorkflowsFor(Segment segment, @Nullable TrackingToken claimedFrom, ProcessingContext sourcingContext, ProcessingContext executionContext)
      Restores and starts the workflow executions owned by a segment this node just claimed.

      Restoration is per segment rather than per node: a node materializes only the instances of the segments it holds, so a segment migrating between nodes carries its instances with it. When another node dies, the coordinator hands its segments to a surviving node and this callback rebuilds their instances there, without restarting anything.

      The instances are materialized from their current durable state and started immediately.

      Parameters:
      segment - the segment that was claimed
      claimedFrom - the stored position the segment resumes from, or null when it has consumed nothing. This is what tells a segment sitting at the end of the stream from one far behind, on a node that never held the segment as well as on one that did
      sourcingContext - short-lived context used to load durable workflow state
      executionContext - independent context used as the parent of the restored workflow executions. Reusing sourcingContext here is invalid because restored workflow bodies run asynchronously and can outlive the claim callback. If such a body appends an event after the sourcing unit of work has entered COMMIT, Axon can no longer register the required PREPARE_COMMIT handler, and the workflow cannot persist its resumed, timed-out, or terminal state
      Returns:
      a future that completes once the segment's executions are restored and, unless the segment is still replaying, started. It completes exceptionally when loading durable state does not finish within DEFAULT_RESTORE_TIMEOUT, failing the claim rather than holding the segment

    • releaseWorkflowsFor

      public CompletableFuture<Void> releaseWorkflowsFor(Segment segment)
      Drops the workflow executions owned by a segment this node no longer holds.

      In-flight steps are stopped rather than cancelled, exactly as on shutdown(): no cancellation events are emitted, so the node claiming the segment next resumes each instance from its persisted state.

      Parameters:
      segment - the segment that was released
      Returns:
      a future completed after this engine has requested every released execution to stop

    • shutdown

      public void shutdown()
      Shuts downs the engine and removes all running workflow executions.

      Before clearing the repository, all in-flight step futures are interrupted so that workflow driver threads parked in sleepAsync / waitForEvent / async execute can exit. This is an interrupt, not a cancellation: no <Step>Cancelled / <Workflow>Cancelled events are emitted, so the persisted event stream still reflects the most recent <Step>Started and the step resumes on the next app start. Without this, a graceful shutdown can hang because the workflow executor (e.g. a virtual-thread-per-task executor) blocks on close() waiting for those threads to terminate. See issue #125.

      A workflow driver that exits after this call skips its termination handler, so it touches no engine state and requests no checkpoint.

    • segmentedRouting

      public SequencingPolicy<EventMessage> segmentedRouting()
      Returns the sequencing policy derived from this engine's workflow definitions, which delivers each event to the segment owning the workflow instance it affects.
      Returns:
      the sequencing policy of this engine.