Message replays

Fluxzero allows you to replay past messages by tracking from an earlier index in the message log. This is useful for:

• Rebuilding projections or read models
• Repairing missed or failed messages
• Retrospectively introducing new functionality

Since message logs — including commands, events, errors, etc. — are durably stored (with events retained indefinitely by default), replays are usually available.

Replaying from the Past

The most common way to initiate a replay is to define a new consumer using the @Consumer annotation with a minIndex.

Java

@Consumer(name = "auditReplay", minIndex = 111677748019200000L)
public class AuditReplayHandler {

    @HandleEvent
    void on(CreateUser event) {
        // ... rebuild projections, audit, etc.
    }
}

Kotlin

@Consumer(name = "auditReplay", minIndex = 111677748019200000L)
class AuditReplayHandler {

    @HandleEvent
    fun on(event: CreateUser) {
        // ... rebuild projections, audit, etc.
    }
}

Note

minIndex is the inclusive starting position in the log. You can derive it from a timestamp using IndexUtils.

Java

long index = IndexUtils.indexFromTimestamp(
    Instant.parse("2024-01-01T00:00:00Z"));
// -> 111677748019200000L

Kotlin

val index = IndexUtils.indexFromTimestamp(
    Instant.parse("2024-01-01T00:00:00Z"))
// -> 111677748019200000L

This approach is perfect for:

• Starting fresh consumers for replays
• Bootstrapping projections without interfering with live handlers
• Keeping logic encapsulated and isolated

Resetting Existing Consumers

If you want to reset an existing consumer to an earlier point in the log:

Java

long replayIndex = 111677748019200000L;

Fluxzero.client()
    .getTrackingClient(MessageType.EVENT)
    .resetPosition("myConsumer", replayIndex, Guarantee.STORED);

Kotlin

val replayIndex = 111677748019200000L

Fluxzero.client()
    .getTrackingClient(MessageType.EVENT)
    .resetPosition("myConsumer", replayIndex, Guarantee.STORED)

Caution

Be careful when replaying on existing handlers to avoid duplicating side effects (emails, HTTP calls, etc.). Prefer a dedicated replay consumer when side effects are not replay-safe.

Parallel replays with exclusive = false

Sometimes you want the same handler class to do both live processing and a historical replay (e.g., to rebuild read models).

Mark the handler’s consumer as exclusive = false:

Java

@Consumer(name = "live", exclusive = false)
public class OrderProcessor {

    @HandleCommand
    void handle(SendOrder command) {
        // submit an order
    }
}

Kotlin

@Consumer(name = "live", exclusive = false)
class OrderProcessor {

    @HandleCommand
    fun handle(command: SendOrder) {
        // submit an order
    }
}

Then register a second consumer at runtime that targets the same handler but a different index window:

Java

fluxzeroBuilder.addConsumerConfiguration(
    ConsumerConfiguration.builder()
        .name("replay")                                  // new consumer
        .handlerFilter(h -> h instanceof OrderProcessor) // same class
        .minIndex(111677748019200000L)                   // start of window
        .maxIndexExclusive(111853279641600000L)          // end of window (exclusive)
        .build(),
    MessageType.COMMAND
);

Kotlin

fluxzeroBuilder.addConsumerConfiguration(
    ConsumerConfiguration.builder()
        .name("replay")                             // new consumer
        .handlerFilter { it is OrderProcessor }     // same class
        .minIndex(111677748019200000L)              // start of window
        .maxIndexExclusive(111853279641600000L)     // end of window (exclusive)
        .build(),
    MessageType.COMMAND
)

This spins up a parallel tracker that replays messages between the two indices while the original live consumer continues uninterrupted.

Tip

Great for bootstrapping read models or repair jobs without disrupting production flow or duplicating classes.

---

Handling failures and DLQ flows

For replaying failures from the error log (dynamic DLQ), use the dedicated Dynamic dead-lettering guide.