Skip to main content
StreamMetricsProvider is the SPI through which FlowWarden Stream Core emits metrics events — stream lifecycle, event flow, errors, checkpoints, DLQ writes, leadership changes. Implementations are fire-and-forget: they must never throw, never block the calling thread. The default implementation is a no-op. The FlowWarden Reporter is the canonical implementation — it aggregates the callbacks and forwards them either to the FlowWarden Console (heartbeat-style protobuf) or to a Micrometer registry (Prometheus). Custom implementations are wired by calling FlowWardenMetrics.setProvider(...) at startup.

​
How wiring works

Unlike LockService / CheckpointStore / DlqStore, this SPI is not resolved as a Spring bean. The framework reads the active provider through a static registry: 
public final class FlowWardenMetrics {
    public static StreamMetricsProvider get();
    public static void setProvider(StreamMetricsProvider provider);
}
A provider library registers itself once during application startup: 
@PostConstruct
void register() {
    FlowWardenMetrics.setProvider(new MyMetricsProvider(...));
}
The default value is StreamMetricsProvider.noOp(). If no provider registers, every callback is silently dropped — Stream Core has no overhead and no behavioural dependency on metrics being collected.

​
Callback contract

All callbacks must not throw and must not block. A throwing or blocking provider can corrupt the calling stream — the hot-path code paths that invoke these callbacks (event delivery, checkpoint writes, DLQ writes) do not wrap them in try / catch. Aggregate state in memory; defer remote calls and disk I/O to a background queue.
The default methods on the interface are {} empty so an implementation only overrides the callbacks it cares about — adding a new optional callback in a future version of FlowWarden does not break existing providers.

​
Lifecycle callbacks

void onStreamStarted(String streamName, StreamConfiguration config);

default void onStreamStopped(String streamName, StopReason reason, Throwable cause) {}
onStreamStopped is emitted in three situations, captured by StopReason:
StopReasonWhencause
GRACEFULExplicit stopStream(), JVM shutdown, leadership loss.null
CRASHEDUncaught Throwable killed the imperative listener container, or the reactive pipeline terminated on an unexpected onError / onComplete.The escaping throwable (or null if it could not be captured)
The CRASHED signal exists so monitoring can distinguish a silently dead stream from an intentional stop — without it, dashboards would report RUNNING indefinitely. Always handle CRASHED explicitly in custom providers; treating it identically to GRACEFUL masks real outages.

​
Event flow callbacks

void onEventReceived(String streamName, ChangeEventMetadata metadata);

void onEventProcessed(String streamName, long durationNanos, boolean success);

void onEventError(String streamName, Throwable error, boolean willRetry,
                  int attemptNumber, ChangeEventMetadata metadata);
  • onEventReceived is called when an event arrives from MongoDB, before any handler runs.
  • onEventProcessed is called after the handler returns — success=false if it threw and success=true if it returned cleanly (after all configured retries succeeded).
  • onEventError is called when a handler throws. willRetry=true means the framework will retry (under the configured RetryPolicy); willRetry=false means the budget is exhausted and the event will move to DLQ or be dropped.

​
Buffer & backpressure callbacks

void onBufferStatus(String streamName, int currentSize, int maxSize);
void onBackpressure(String streamName, BackpressureAction action);
onBufferStatus is invoked when the internal event buffer size changes meaningfully (the framework batches these — not one call per event). onBackpressure fires when an action from BackpressureAction is triggered (e.g. drop-oldest, throttle).

​
Checkpoint callbacks

void onCheckpoint(String streamName, String resumeToken);

default void onCheckpointFailed(String streamName, Throwable cause) {}

default void onResumeFallbackToSeen(String streamName) {}
default void onResumeFallbackToProcessed(String streamName) {}
default void onResumeHistoryLost(String streamName) {}

default void onCheckpointLag(String streamName, long lagSeconds, long lagEvents) {}
  • onCheckpoint is emitted only on confirmed write success from the CheckpointStore. If the store throws, this is not invoked — onCheckpointFailed is, with the thrown cause. Stream-core does not stop the stream on a checkpoint write failure; it logs the failure, emits the signal, and keeps processing.
  • The three resume callbacks signal level-2 and level-3 events in the resume cascade — see @Checkpoint resume cascade for the cascade rules. They are graceful-degradation signals (the stream continues; the operator may want to investigate why a level-2 or level-3 fallback happened).
  • onCheckpointLag is called periodically with the current lag between lastSeenToken and lastProcessedToken — useful for monitoring streams whose handlers fall behind.

​
DLQ callbacks

void onEventSentToDlq(String streamName);
default void onEventDlqFailed(String streamName, Throwable cause) {}
  • onEventSentToDlq is emitted only on confirmed write success from the DlqStore.
  • onEventDlqFailed fires when the store throws. A DLQ write failure is operationally critical: the event has neither been processed by the handler nor archived for later replay. Stream-core logs and emits the signal; the auto post-retry path then resumes the stream, while manual ctx.sendToDlq(...) paths re-throw to the user handler. Wire alerts on this signal.

​
Infrastructure & leadership callbacks

void onOplogStats(double logLengthHours, String status);

default void onLeadershipChange(String streamName, String role, String instanceId) {}
  • onOplogStats reports the MongoDB oplog window size. status is "OK", "UNAVAILABLE" (collection inaccessible — e.g. Atlas shared tier), or "ERROR". A shrinking window matters for resume safety — a short window relative to your downtime means onHistoryLost is closer than you think.
  • onLeadershipChange fires on SINGLE_LEADER streams when the local instance’s role changes. role is "LEADER", "STANDBY", or "NOT_APPLICABLE" (for ALL_INSTANCES streams).

​
Provided implementations

ImplementationModuleWhat it does
NoOpStreamMetricsProviderflowwarden-stream-coreDefault. Silently discards all callbacks. Active until something calls setProvider.
FlowWarden Reporterflowwarden-reporterAggregates callbacks per stream; ships them to the FlowWarden Console (protobuf heartbeat) and/or to a Micrometer registry. The canonical reference implementation.
No contract test ships in flowwarden-stream-core-testkit for this SPI — the surface is large and the contract is “don’t throw, don’t block” rather than algorithmic. The Reporter is validated through its own integration tests against the protobuf wire format and the Micrometer registry.

​
Providing a custom implementation

Implement the interface, override only the callbacks you care about, and register at startup: 
@Component
public class MyMetricsProvider implements StreamMetricsProvider {

    private final MyAggregator aggregator;

    @PostConstruct
    void register() {
        FlowWardenMetrics.setProvider(this);
    }

    @Override
    public void onEventReceived(String streamName, ChangeEventMetadata metadata) {
        aggregator.recordReceived(streamName);
    }

    @Override
    public void onEventError(String streamName, Throwable error, boolean willRetry,
                             int attemptNumber, ChangeEventMetadata metadata) {
        aggregator.recordError(streamName, willRetry);
    }

    // ... override only what you need ...
}
Because callbacks are fire-and-forget, the framework will never observe whether your provider succeeded — but it also won’t shield other listeners from a misbehaving one. Keep work in-memory and offload to a background queue.

​
See Also

FlowWarden Reporter

The canonical implementation: Console + Prometheus modes.

@Checkpoint resume cascade

Context for the three onResumeFallback* callbacks.

CheckpointStore SPI

The store whose write success / failure feeds onCheckpoint*.

DlqStore SPI

The store whose write success / failure feeds onEventSentToDlq / onEventDlqFailed.