@RestartPolicy

The @RestartPolicy annotation (since 1.0.0-rc.7) tunes the managed restart loop of a stream: how long FlowWarden waits between resubscription attempts after the stream died at runtime, and whether it ever gives up. Place it on a class alongside @ChangeStream. Without the annotation, every stream uses the defaults below — which are exactly the behaviour FlowWarden had before the annotation existed. Nothing changes for an application that does not use it.

What the restart loop does

A stream can die at runtime, after it started successfully: the cursor is closed by a network outage or a primary stepdown, the server returns a non-resumable error, or the watched collection is dropped or renamed (an invalidate event). In every case FlowWarden:

evicts the stream’s state and reports it stopped — onStreamStopped(CRASHED) on the metrics provider;
hands the stream to the restart loop, which re-enters the full startup path (resume cascade included) after a backoff delay;
on success, reports onStreamRestarted(streamName, attempt, cause) and resets the attempt counter;
on a transient failure (server still down, cascade probe failure), schedules the next attempt with a longer delay.

A HistoryLostException escaping an attempt — the resume cascade escalated to onHistoryLost = FAIL — is always terminal: the loop stops for that stream, a second CRASHED signal carries the exception, and under SINGLE_LEADER the lease is released so a standby can take over. A manual startStream (actuator or FlowWarden API) during the backoff wins over the loop: the installed stream is left alone. Restart attempts run on one dedicated thread per manager (fw-stream-restart / fw-stream-restart-reactive), so a slow resubscription never delays heartbeat probes or checkpoint flushes of healthy streams.

Attributes

Attribute	Type	Default	Description
`maxAttempts`	`int`	`0`	Resubscription attempts before giving up. `0` = unlimited
`initialDelay`	`String`	`"1s"`	Delay before the first attempt. Supports `"500ms"`, `"1s"`, `"1m"` format. Must be `> 0`
`maxDelay`	`String`	`"60s"`	Maximum delay cap. Must be `> 0`
`multiplier`	`double`	`2.0`	Multiplier applied to the delay between attempts. Must be finite and `>= 1`
`jitter`	`boolean`	`false`	Adds random variation (±20%) to the computed delay, floored at 1ms

Two deliberate differences from @RetryPolicy:

maxAttempts counts resubscription attempts — there is no “initial attempt” to include — and 0 means unlimited, which is the default. @RetryPolicy.maxAttempts must be >= 1.
jitter defaults to false, so a stream without the annotation keeps the deterministic cadence it always had. @RetryPolicy.jitter defaults to true.

Backoff Algorithm

Same shape as @RetryPolicy:

baseDelay   = initialDelay × (multiplier ^ (attemptNumber - 1))
cappedDelay = min(baseDelay, maxDelay)

if jitter = true:
    finalDelay = cappedDelay × random(0.8, 1.2)    // ±20%
else:
    finalDelay = cappedDelay

With the defaults the delays are:

Attempt	Delay before it
1	1s
2	2s
3	4s
4	8s
5	16s
6	32s
7+	60s (capped)

The counter resets on the first successful resubscription: a stream that dies again later starts over at attempt 1.

Giving up

When maxAttempts is set and exhausted, the stream takes the same terminal path as a history loss:

the stream is reported stopped — onStreamStopped(CRASHED) with the original death cause;
the loop stops for that stream — no further attempts, no further WARN lines;
under SINGLE_LEADER, the lease is released so another instance can take over (and surface the same failure if the cause is global);
the stream stays stopped until an operator restarts it — POST /actuator/flowwarden/{streamName}/restart or FlowWarden.startStream(..).

The give-up is immediate once the ceiling is crossed — it does not wait for another backoff.

Every death notification consumes one attempt, not only failed startStream calls. In particular, a reactive subscription that terminates synchronously right after a successful resubscription re-arms the loop and counts as a consumed attempt.

When to set it

Situation	Suggested policy
Default — database outages must never leave streams dead once it comes back	Leave the annotation off (unlimited, 1s → 60s)
A stream that can be permanently broken for a non-history reason — rotated credentials behind a `mongoTemplateRef`, a cluster that will not return — and should stop logging every minute forever	`maxAttempts = 30` (about 25 minutes of attempts with the default cadence), then rely on the `CRASHED` signal and an operator
Latency-sensitive stream on a cluster with short blips	Lower cap: `maxDelay = "10s"`
Many streams on one shared cluster, all resubscribing in the same window when it returns	`jitter = true`, possibly a larger `maxDelay`

@ChangeStream(collection = "orders", mongoTemplateRef = "ordersTemplate")
@RestartPolicy(maxAttempts = 30, initialDelay = "2s", maxDelay = "2m", jitter = true)
public class OrderProjection {

    @OnChange
    public void onChange(ChangeStreamContext<Order> ctx) {
        // ...
    }
}

Validation

Validated fail-fast at startup, on both the annotation path and the registration API:

maxAttempts >= 0
multiplier finite and >= 1 — a shrinking cadence would degenerate into a hot loop against a down server; NaN and infinities are rejected rather than silently collapsing to maxDelay
initialDelay and maxDelay parse (<number>ms|s|m) and are strictly positive — same reason

A violation fails application startup with a BeanCreationException naming the attribute.

Programmatic equivalent

RestartPolicySpec on StreamSpec.Builder.restartPolicy(..), same defaults:

registration.stream("order-stream", Order.class)
        .collection("orders")
        .restartPolicy(RestartPolicySpec.builder()
                .maxAttempts(30)
                .maxDelay("2m")
                .jitter(true)
                .build())
        .onChange(ctx -> { /* ... */ });

@RetryPolicy

Retry of a failed handler invocation — per event, not per stream

@Checkpoint

Resume cascade and the onHistoryLost terminal path the loop shares

StreamMetricsProvider

onStreamStopped(CRASHED) and onStreamRestarted signals

StreamSpec

RestartPolicySpec for contributed streams

Annotations

Registration API

Runtime

SPI

What the restart loop does

Attributes

Backoff Algorithm

Giving up

When to set it

Validation

Programmatic equivalent

See Also

@RetryPolicy

@Checkpoint

StreamMetricsProvider

StreamSpec

​What the restart loop does

​Attributes

​Backoff Algorithm

​Giving up

​When to set it

​Validation

​Programmatic equivalent

​See Also

@RetryPolicy

@Checkpoint

StreamMetricsProvider

StreamSpec

What the restart loop does

Attributes

Backoff Algorithm

Giving up

When to set it

Validation

Programmatic equivalent

See Also