> ## Documentation Index
> Fetch the complete documentation index at: https://docs.flowwarden.io/llms.txt
> Use this file to discover all available pages before exploring further.
# StreamMetricsProvider SPI
> Fire-and-forget callbacks for collecting metrics from FlowWarden Change Streams
`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](/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:
```java theme={null}
public final class FlowWardenMetrics {
public static StreamMetricsProvider get();
public static void setProvider(StreamMetricsProvider provider);
}
```
A provider library registers itself once during application startup:
```java theme={null}
@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
```java theme={null}
void onStreamStarted(String streamName, StreamConfiguration config);
default void onStreamStopped(String streamName, StopReason reason, Throwable cause) {}
```
`onStreamStopped` is emitted in three situations, captured by `StopReason`:
| `StopReason` | When | `cause` |
| ------------ | --------------------------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------- |
| `GRACEFUL` | Explicit `stopStream()`, JVM shutdown, leadership loss. | `null` |
| `CRASHED` | Uncaught `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
```java theme={null}
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
```java theme={null}
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
```java theme={null}
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](/reference/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
```java theme={null}
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
```java theme={null}
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
| Implementation | Module | What it does |
| --------------------------- | ---------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `NoOpStreamMetricsProvider` | `flowwarden-stream-core` | Default. Silently discards all callbacks. Active until something calls `setProvider`. |
| FlowWarden Reporter | [`flowwarden-reporter`](/reporter) | Aggregates 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:
```java theme={null}
@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
The canonical implementation: Console + Prometheus modes.
Context for the three `onResumeFallback*` callbacks.
The store whose write success / failure feeds `onCheckpoint*`.
The store whose write success / failure feeds `onEventSentToDlq` / `onEventDlqFailed`.