Package io.axoniq.framework.messaging.eventstreaming.checkpoint

Interface Checkpointing

@Internal public interface Checkpointing
Declares that an event-handling unit manages its own progress: it decides when its work for a Segment is durable and asks the processor to advance the stored TrackingToken through a CheckpointTrigger. Implement this when handling writes asynchronously, or triggers work that is not durable by the time the event handler returns.

Two kinds of unit implement it: an annotated handler POJO (whose wrapping AnnotatedEventHandlingComponent detects it and exposes it via EventHandlingComponent.unwrap(Class)), and a programmatic component (which implements both EventHandlingComponent and this interface). Either way the processor finds it via unwrap(Checkpointing.class) and runs in request-driven mode; a processor whose every handler is Checkpointing is fully-deferred (it advances the stored token only on explicit request), whereas a processor with at least one ordinary handler runs in auto mode (it requests a checkpoint at the batch-end token every batch).

Only a streaming processor (with a tracking token and segments) can honour this protocol. A processor that does not stream (such as a SubscribingEventProcessor) simply never invokes the lifecycle callbacks (no onSegmentClaimed(Segment, CheckpointTrigger), no checkpoints) and does not expose a CheckpointTrigger, so the checkpointing behaviour is inert there.

A checkpointing unit may use one of two styles:

Both are valid because onCheckpointAdvanced(Segment, TrackingToken) returns a future (carrying the reached token) the processor awaits before storing.

Orthogonal to when the request is made is the kind of workload the future represents. Two patterns are common, and both rely on the same future-based contract:

  • In-memory projection with asynchronous persistence: the unit applies each event to in-memory state on the processing thread (so ordering is naturally preserved) and only persists asynchronously. When a checkpoint is taken it snapshots that state to durable storage and completes the future once the snapshot is durable, reporting the snapshotted position. Event handling resumes as soon as the future completes; the future merely gates storing the token on the snapshot becoming durable.
  • Truly asynchronous handlers: the unit dispatches work that runs independently of event handling and does not block the next event, so it may fall behind the stream. Ordering across that work is then the unit's own responsibility, but it has full control over it. At checkpoint time the unit returns a future that completes only once the in-flight work has caught up far enough to cover the requested position. Because the processor awaits that future before resuming the segment's worker, the checkpoint effectively becomes a barrier: event handling is "blocked" until the asynchronous work has reached the checkpoint.

The processor awaits the returned future without a timeout: it stores the checkpoint (and, on release, frees the claim) only once the future completes. A future that never completes therefore stalls progress for that segment indefinitely. The framework deliberately does not impose a timeout: a unit that confirms durability out of band may legitimately take a long time. If a unit's asynchronous work could hang, it should bound its own future, for example: 


 public CompletableFuture<TrackingToken> onCheckpointAdvanced(Segment segment, TrackingToken requested) {
     return flushToStore(requested)            // returns CompletableFuture<TrackingToken>
             .orTimeout(30, TimeUnit.SECONDS);  // fails the checkpoint instead of stalling the segment forever
 }
A future completed exceptionally (whether from such a timeout or any other failure) fails the checkpoint without storing, leaving the stored token where it is; the segment retries on the next cycle.

Transactional coupling within a batch. The checkpoint is taken on the commit of the same transaction that processed the batch: the processor awaits the returned future while that transaction is still open. Two consequences follow. First, a slow or never-completing future does not merely stall progress: it holds the batch's transaction open for the duration of the await. Second, if the checkpoint fails (the future completes exceptionally), the batch transaction rolls back, undoing the work of every handler in that batch, including ordinary (non-checkpointing) handlers and other checkpointing components sharing the segment. A misbehaving checkpointing component can therefore affect co-located handlers. For this reason, a checkpointing component is best isolated in its own pooled streaming event processor (one whose every handler is Checkpointing, so it runs fully-deferred) rather than mixed with ordinary handlers, both to avoid the rollback coupling and because in a mixed processor auto checkpointing wins and the component cannot actually defer its segment's token.

The only method that must be implemented is onCheckpointAdvanced(Segment, TrackingToken); onSegmentClaimed(Segment, CheckpointTrigger) and onSegmentReleased(Segment, TrackingToken) have sensible defaults (see each method).

Internal API. This interface is marked Internal: self-checkpointing is currently intended primarily for internal and advanced use, is not part of the documented public feature set, and its shape may change in a minor or patch release.

Since:
5.3.0
Author:
Allard Buijze
See Also:

  • Method Details

    • onSegmentClaimed

      default void onSegmentClaimed(Segment segment, CheckpointTrigger trigger)
      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 onSegmentReleased(Segment, TrackingToken).

      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.

      Parameters:
      segment - the segment that was claimed
      trigger - the handle to request checkpoints for segment

    • onCheckpointAdvanced

      CompletableFuture<TrackingToken> onCheckpointAdvanced(Segment segment, TrackingToken requested)
      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.

      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

      default CompletableFuture<TrackingToken> onSegmentReleased(Segment segment, TrackingToken upTo)
      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 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 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.

      Parameters:
      segment - the segment being released
      upTo - 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