@Checkpoint

​

Basic Usage

@ChangeStream(name = "order-watcher", collection = "orders")
@Checkpoint
public class OrderStreamHandler {

    @OnInsert
    void handle(ChangeStreamContext<Order> ctx) {
        System.out.println(ctx.summary());
    }
}

​

Attributes

​

saveEveryN

// Anchor every 10 settlements
@Checkpoint(saveEveryN = 10)

​

saveIntervalSeconds

// A dirty anchor is persisted within 30 seconds
@Checkpoint(saveIntervalSeconds = 30)

// Count threshold only
@Checkpoint(saveIntervalSeconds = 0)

​

idleHeartbeatIntervalSeconds

// Aggressive protection for a short oplog window
@Checkpoint(idleHeartbeatIntervalSeconds = 60)

// Opt out — idle streams are then exposed to oplog rollover
@Checkpoint(idleHeartbeatIntervalSeconds = 0)

​

stuckAfterProbes

• only abstentions with no delivery in between count — any settled event resets the streak;
• only idle probes count — the startup diagnostic and recovery probes never do;
• a stream with a handler in flight is never declared stuck. A slow handler or a retry in backoff holds the settled position still exactly like a dead cursor would, and must not be restarted mid-handler; its settlement resets the streak, a dead cursor’s never will.

// Faster verdict on a latency-sensitive stream: 2 × 60s
@Checkpoint(idleHeartbeatIntervalSeconds = 60, stuckAfterProbes = 2)

// Keep the WARN, never restart on the probe's verdict
@Checkpoint(stuckAfterProbes = 0)

​

startPosition

​

Resume cascade

​

onHistoryLost

// Default: fail loudly — operator must decide
@Checkpoint(onHistoryLost = OnHistoryLost.FAIL)

// Resume from oldest available oplog entry (partial recovery)
@Checkpoint(onHistoryLost = OnHistoryLost.RESUME_FROM_OPLOG_START)

// Skip gap entirely — for streams where missing events is acceptable
@Checkpoint(onHistoryLost = OnHistoryLost.RESUME_FROM_NOW)

public class HistoryLostException extends RuntimeException {
    public String getStreamName();
    public Instant getLastCheckpointTimestamp();
}

​

Comprehensive Example

@ChangeStream(name = "order-stream", collection = "orders", documentType = Order.class)
@Checkpoint(saveEveryN = 5, saveIntervalSeconds = 10, startPosition = StartPosition.LATEST)
public class OrderStreamHandler {

    @OnInsert
    void handle(ChangeStreamContext<Order> ctx) {
        Order order = ctx.getFullDocument(Order.class);
        log.info("New order: {}", order.getId());
    }
}

@ChangeStream(name = "order-stream", collection = "orders", documentType = Order.class)
@Checkpoint(saveEveryN = 5, saveIntervalSeconds = 10, startPosition = StartPosition.LATEST)
public class OrderStreamHandler {

    @OnInsert
    Mono<Void> handle(ChangeStreamContext<Order> ctx) {
        Order order = ctx.getFullDocument(Order.class);
        log.info("New order: {}", order.getId());
        return Mono.empty();
    }
}

​

How It Works

​

Storage

​

Best Practices

• Keep saveEveryN = 1 for critical streams (e.g., order processing, billing) to minimize event replay on crash.
• Increase saveEveryN for high-throughput streams where replaying a few events is acceptable — saveIntervalSeconds still bounds the anchor’s age in time.
• Do not disable idleHeartbeatIntervalSeconds unless you have a specific reason. It is the only mechanism that keeps an idle stream’s resume point inside the oplog window.
• Use startPosition = LATEST only for streams where historical events are irrelevant (e.g., cache invalidation).
• Alert on the age of lastHeartbeatTimestamp — it records the last confirmed recoverable position (anchor write or certification) and never moves on probe abstentions: a frozen value signals a stuck stream or a failing store.
• Alert on the onResumeFallbackToSeen callback (surfaced by the Reporter as a resume outcome, or by your own StreamMetricsProvider) — a recurring level-2 fallback on an active stream signals that the processed anchor ages out faster than the stream settles, typically a stuck or retry-pathological handler.
• Alert on the onResumeHistoryLost callback — reaching level 3 means downtime overran the oplog window with no usable fallback; investigate operationally rather than silencing it.

​

See Also