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

Class WorkflowEngineCheckpointingSupport

java.lang.Object

io.axoniq.framework.workflow.runtime.execution.WorkflowEngineCheckpointingSupport

All Implemented Interfaces:

Checkpointing

@Internal
public class WorkflowEngineCheckpointingSupport
extends Object
implements Checkpointing

A Checkpointing implementation dedicated for a WorkflowEngine, to separate check pointing behavior from it.

This class has one job: coordinate checkpoint requests and complete onCheckpointAdvanced(Segment, TrackingToken) only after the host confirms that workflow-owned asynchronous work is safe. It does not track processor progress or decide when the engine should switch to live mode.

Everything here is kept per segment. The processor hands out one CheckpointTrigger per claimed segment, a request through it advances only that segment's stored token, and it is inert once that claim ends.

The trigger map is concurrent, and its updates are deliberately small: they never invoke processor or workflow callbacks while a claim is being recorded or dropped. The coordinator callback supplied to onCheckpointAdvanced(Segment, TrackingToken) re-enters the support only after all the WorkflowExecution latches (attached by the WorkflowEngineCheckpointingSupport.CheckpointLatchCoordinator) have been crossed.

Since:

5.4.0

Author:

Simon Zambrovski, Steven van Beelen

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static interface	WorkflowEngineCheckpointingSupport.CheckpointLatchCoordinator	Coordinates workflow work that must complete before checkpoint advancement.

Constructor Summary

Constructors

Constructor	Description
WorkflowEngineCheckpointingSupport(WorkflowEngineCheckpointingSupport.CheckpointLatchCoordinator checkpointLatchCoordinator)	Creates checkpointing support for a WorkflowEngine using the given checkpointLatchCoordinator.

Method Summary

Modifier and Type	Method	Description
CompletableFuture<TrackingToken>	onCheckpointAdvanced(Segment segment, TrackingToken requested)	Invoked on the processing thread when the processor takes a checkpoint for segment that must cover requested.
void	onSegmentClaimed(Segment segment, @Nullable TrackingToken from, CheckpointTrigger trigger)	Invoked when segment is claimed, handing the unit the CheckpointTrigger it uses to declare safe positions for that segment.
CompletableFuture<TrackingToken>	onSegmentReleased(Segment segment, TrackingToken requested)	Invoked when segment is being released.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Details

WorkflowEngineCheckpointingSupport

public WorkflowEngineCheckpointingSupport(WorkflowEngineCheckpointingSupport.CheckpointLatchCoordinator checkpointLatchCoordinator)

Creates checkpointing support for a WorkflowEngine using the given checkpointLatchCoordinator.

Parameters:

checkpointLatchCoordinator - adds a latch to coordinated WorkflowExecutions to ensure all have reached a safe point to advance the checkpoint

Method Details

onSegmentClaimed

public void onSegmentClaimed(Segment segment,
 @Nullable TrackingToken from,
 CheckpointTrigger trigger)

Description copied from interface: Checkpointing

Invoked when segment is claimed, handing the unit the CheckpointTrigger it uses to declare safe positions for that segment. Retain it keyed by segment; it is invalid after Checkpointing.onSegmentReleased(Segment, TrackingToken).

The from token is the position this segment resumes at, the same position Checkpointing.onSegmentReleased(Segment, TrackingToken) carries. It shows how far along the stream the segment is before the first event is delivered, so a unit restoring state at claim time can see whether it is picking up a segment at the end of the stream or one that still has history to replay.

Defaults to a no-op: a unit that obtains its trigger another way (typically through a CheckpointTrigger handler-method parameter) does not need to retain it here.

Invoked synchronously as part of claiming the segment. Throwing from this method fails the claim cycle: the segment is not claimed and the coordinator retries it (after a back-off). This is deliberate: a component that never received its trigger could never checkpoint, so signal a genuine inability to accept the claim by throwing, but do not throw for transient conditions that a retained trigger would handle later.

Specified by:

onSegmentClaimed in interface Checkpointing

Parameters:

segment - the segment that was claimed

from - the segment's stored TrackingToken, the position processing resumes from, or null when processing starts at the beginning of the stream

trigger - the handle to request checkpoints for segment

onCheckpointAdvanced

public CompletableFuture<TrackingToken> onCheckpointAdvanced(Segment segment,
 TrackingToken requested)

Description copied from interface: Checkpointing

Invoked on the processing thread when the processor takes a checkpoint for segment that must cover requested. Ensure work is durable at least up to requested, discard buffered state, then return the highest token now safe, which may exceed requested if more async work has drained. The processor stores the reported token only after every returned future completes. When several checkpointing units share a segment, it does not simply store the lowest report: it reconciles their positions to a single agreed token (the highest any unit reported), re-requesting any unit that has not yet reached it, and stores that, so no unit is ever left durably ahead of the stored token.

Contract: the returned token must cover requested. The processor only ever advances the stored token: a value at or behind the last stored checkpoint is ignored (never rewinds progress).

Exceptional completion. A unit that cannot reach requested must complete its future exceptionally (returning a token that does not cover requested is treated the same way). The checkpoint then fails and nothing is stored. The processor handles this as a processing error: it aborts the segment's worker and releases the segment, so the coordinator re-claims it (after a back-off) and resumes from the last stored checkpoint; the events since are redelivered. The stored token therefore never advances on a failed checkpoint, and no progress is lost; an isolated failure costs a segment re-claim and some redelivery.

The returned future is the lever for the two workloads described on this type. For an in-memory projection with asynchronous persistence, complete the future once the snapshot of the in-memory state has become durable, reporting the snapshotted position; the processing thread then resumes immediately. For a truly asynchronous handler, complete the future only once the in-flight work has caught up far enough to cover requested; the processor awaits it before resuming the segment's worker, so the checkpoint acts as a barrier that holds event handling until the asynchronous work has reached this position.

Specified by:

onCheckpointAdvanced in interface Checkpointing

Parameters:

segment - the segment being checkpointed

requested - the segment-scoped position this checkpoint must cover (the highest anyone requested, or the batch-end token in auto mode)

Returns:

the highest token this unit is durably safe at; must cover requested and must not sit behind the last persisted checkpoint

onSegmentReleased

public CompletableFuture<TrackingToken> onSegmentReleased(Segment segment,
 TrackingToken requested)

Description copied from interface: Checkpointing

Invoked when segment is being released. Flush everything handled so far, discard the segment's local state, and report the highest token made durable; the processor reconciles it with any co-located units (the same way as Checkpointing.onCheckpointAdvanced(Segment, TrackingToken)) and stores the result while still holding the claim, then releases the claim. After this the CheckpointTrigger for the segment is invalid.

The returned token must lie in the range [lastStoredCheckpoint, upTo]: it must not rewind below the last persisted checkpoint (a regressive value is ignored), and it cannot exceed upTo (a unit can only be durable up to what it was handed). Returning a token below upTo is the normal lagging case: the uncovered tail (from the returned token up to upTo) is simply reprocessed on the next claim. Forcing the value all the way up to upTo is not required and is often impossible for asynchronous work.

The same two workloads described on this type apply here. An in-memory projection with asynchronous persistence takes a final snapshot and completes the future once it is durable, normally reporting upTo. A truly asynchronous handler completes the future once its in-flight work has drained as far as it can before the claim is given up, reporting whatever position is durable by then, typically lower than upTo, with the uncovered tail reprocessed on the next claim. Unlike onCheckpointAdvanced, reaching upTo is not required here: there is no later request to satisfy, so an asynchronous unit may report best-effort progress rather than forcing a full drain.

Exceptional completion. Unlike Checkpointing.onCheckpointAdvanced(Segment, TrackingToken), a future that completes exceptionally here does not abort or retry: the segment is being given up regardless. The release still proceeds: the checkpoint simply does not advance for what the unit could not confirm, the claim is freed, and the uncovered tail (everything past the last stored checkpoint) is reprocessed when the segment is next claimed. There is nothing the framework can do about a genuine durability failure at release beyond this reprocessing, so a unit that knows a lower already-durable position should report it (return it) rather than fail, to keep the redelivered window as small as possible.

Defaults to onCheckpointAdvanced(segment, upTo), i.e. a full flush up to the consumed position. Because that delegate must cover its argument or complete exceptionally, the default does not advance the checkpoint on release if the unit cannot reach upTo (the claim is still released, per the exceptional-completion behaviour above); override this method to instead report a lower already-durable position (the relaxed, best-effort semantics) for an asynchronous unit that may legitimately lag behind upTo at release.

Specified by:

onSegmentReleased in interface Checkpointing

Parameters:

segment - the segment being released

requested - the segment's lastConsumedToken: the position to drain toward

Returns:

the highest token durably persisted; at least the last persisted checkpoint and at most upTo, and typically lower than upTo when asynchronous work is still in flight