Transactions

By default, FlowWarden provides at-least-once delivery: if the application crashes between processing an event and saving the checkpoint, the event is replayed on restart. For most use cases (cache invalidation, notifications, projections), this is sufficient as long as your handler is idempotent. For critical streams where duplicate processing is unacceptable (e.g., billing, payments, inventory), you can achieve exactly-once semantics by wrapping your business writes and checkpoint save in a single MongoDB transaction.

Reading transaction metadata

When the upstream writer commits inside a MongoDB transaction, every change event in that transaction carries the session id (lsid) and transaction number (txnNumber). FlowWarden surfaces them through ChangeStreamContext.getTransactionInfo() as an Optional<TransactionInfo>, empty for standalone writes. No extra configuration is required — just the replica set already needed by Change Streams.

@ChangeStream(documentType = Order.class)
public class OrderAuditHandler {

    @OnInsert
    void onInsert(Order order, ChangeStreamContext<Order> ctx) {
        ctx.getTransactionInfo().ifPresentOrElse(
            txn -> log.info("Order {} part of txn {}", order.getId(), txn.txnNumber()),
            ()  -> log.info("Order {} (standalone)", order.getId()));
    }
}

The reactive twin returns Mono<Void> and wraps the body in Mono.fromRunnable; the ctx.getTransactionInfo() call is identical.

Common use cases

Audit / event grouping — bucket downstream log entries by txnNumber so a single business operation surfaces as one logical record
Atomicity-preserving propagation — collect all events of one upstream transaction before publishing them downstream as a unit (Kafka batch, message envelope, projection rebuild)
Correlation — propagate txnNumber into traces / spans so cross-service correlation matches the source-of-truth transaction boundary

The full record signature is TransactionInfo(BsonDocument lsid, long txnNumber). See ChangeStreamContext → TransactionInfo for the exact contract.

A runnable demo with a generator that mixes transactional + standalone inserts lives in the flowwarden-examples repository — see imperative/10-transactions and reactive/10-transactions.

Prerequisites for exactly-once

MongoDB transactions require:

A replica set (already required for Change Streams)
A MongoTransactionManager Spring bean in your application

@Configuration
public class MongoConfig {

    @Bean
    MongoTransactionManager transactionManager(MongoDatabaseFactory dbFactory) {
        return new MongoTransactionManager(dbFactory);
    }
}

Exactly-once pattern

Use Spring’s @Transactional on your handler method combined with ctx.saveCheckpointNow():

@ChangeStream(documentType = Order.class)
@Checkpoint(saveEveryN = 1)
public class OrderProcessor {

    @Autowired
    private MongoTemplate mongoTemplate;

    @OnInsert
    @Transactional
    void handle(Order order, ChangeStreamContext<Order> ctx) {
        mongoTemplate.save(new ProcessedOrder(order), "processed_orders");
        ctx.saveCheckpointNow();
    }
}

On success: both the business write and the checkpoint are committed atomically. On exception: Spring rolls back the transaction — neither the business write nor the checkpoint are persisted. By default, FlowWarden logs the error and moves on to the next event. The failed event is lost unless you configure retry and DLQ:

@ChangeStream(documentType = Order.class)
@Checkpoint(saveEveryN = 1)
@RetryPolicy(maxAttempts = 3)
@DeadLetterQueue
public class OrderProcessor {

    @Autowired
    private MongoTemplate mongoTemplate;

    @OnInsert
    @Transactional
    void handle(Order order, ChangeStreamContext<Order> ctx) {
        mongoTemplate.save(new ProcessedOrder(order), "processed_orders");
        ctx.saveCheckpointNow();
        // On failure: @RetryPolicy retries in a new transaction,
        // then @DeadLetterQueue captures the event if all retries fail
    }
}

For critical streams using transactions, always combine @RetryPolicy and @DeadLetterQueue to avoid losing events on failure. See the Retry & DLQ guide for configuration details.

How it works

Spring’s @Transactional opens a MongoDB session and binds it to the current thread
All MongoTemplate operations on the same thread automatically use the bound session
ctx.saveCheckpointNow() saves the checkpoint through MongoTemplate — it joins the same transaction
On success, Spring commits both the business write and the checkpoint atomically
On failure, Spring rolls back everything — the event will be redelivered on the next attempt

ctx.saveCheckpointNow() is required inside the handler. Without it, FlowWarden saves the checkpoint after the handler returns — outside the transaction boundary.

When to use transactions

Use transactions

All handler writes go to MongoDB
Duplicate processing has real consequences (double charges, wrong inventory)
You need exactly-once guarantees
You use SINGLE_LEADER deployment mode

Skip transactions

Handler calls external APIs (HTTP, Kafka, email)
Handler is naturally idempotent
At-least-once is acceptable
High-throughput streams where transaction overhead matters

MongoDB transactions cannot roll back external side-effects (HTTP calls, message queue publishes, emails). If your handler performs external calls, transactions only protect the MongoDB writes — the side-effects will not be undone on rollback. Design these handlers to be idempotent instead.

Interaction with retry and DLQ

Transactions work naturally with @RetryPolicy and @DeadLetterQueue:

@ChangeStream(documentType = Order.class)
@Checkpoint(saveEveryN = 1)
@RetryPolicy(maxAttempts = 3)
@DeadLetterQueue(retentionDays = 90)
public class CriticalOrderProcessor {

    @Autowired
    private MongoTemplate mongoTemplate;

    @OnInsert
    @Transactional
    void handle(Order order, ChangeStreamContext<Order> ctx) {
        mongoTemplate.save(new ProcessedOrder(order), "processed_orders");
        ctx.saveCheckpointNow();
    }
}

Each retry attempt runs the handler in a new transaction
If the handler throws, the transaction is rolled back (business writes + checkpoint)
After all retries are exhausted, the event is sent to the DLQ outside any transaction

Important constraints

`mongoTemplateRef` breaks transaction atomicity

If your @ChangeStream uses mongoTemplateRef to point to a different MongoTemplate, the exactly-once guarantee does not apply.

Spring binds the transaction session to a specific MongoDatabaseFactory. The checkpoint store uses the default MongoTemplate from auto-configuration. If your handler uses a different template via mongoTemplateRef, the checkpoint save and the handler’s writes use different database factories — they cannot participate in the same transaction.

// This does NOT provide exactly-once:
@ChangeStream(collection = "orders", mongoTemplateRef = "secondaryMongoTemplate")
@Checkpoint(saveEveryN = 1)
public class OrderProcessor {

    @Autowired
    @Qualifier("secondaryMongoTemplate")
    private MongoTemplate secondaryMongoTemplate;

    @OnInsert
    @Transactional  // transaction on secondary factory
    void handle(Order order, ChangeStreamContext<Order> ctx) {
        secondaryMongoTemplate.save(...);  // secondary factory
        ctx.saveCheckpointNow();           // default factory — NOT in the same transaction
    }
}

`saveEveryN` must be 1

For transactional exactly-once, use @Checkpoint(saveEveryN = 1). With saveEveryN > 1, FlowWarden skips checkpoint saves between batches — events processed between checkpoints would not have transactional guarantees on crash recovery.

Handler must call `saveCheckpointNow()`

FlowWarden’s automatic checkpoint save happens after the handler returns — outside the @Transactional boundary. You must explicitly call ctx.saveCheckpointNow() inside the handler to include the checkpoint in the transaction. If you forget, FlowWarden falls back to at-least-once (the checkpoint saves outside the transaction).

Comparison with at-least-once

Aspect	At-least-once (default)	Exactly-once (transactional)
Delivery guarantee	Events may be redelivered on crash	Each event processed exactly once
Handler requirement	Must be idempotent	No idempotency needed (for MongoDB writes)
Performance	No transaction overhead	Slight overhead from MongoDB sessions
External side-effects	Handled naturally	Not protected by transaction
Configuration	`@Checkpoint` only	`@Checkpoint` + `MongoTransactionManager` + `@Transactional` + `saveCheckpointNow()`

Getting Started

FAQ

Guides

Reading transaction metadata

Common use cases

Prerequisites for exactly-once

Exactly-once pattern

How it works

When to use transactions

Use transactions

Skip transactions

Interaction with retry and DLQ

Important constraints

`mongoTemplateRef` breaks transaction atomicity

`saveEveryN` must be 1

Handler must call `saveCheckpointNow()`

Comparison with at-least-once

​Reading transaction metadata

​Common use cases

​Prerequisites for exactly-once

​Exactly-once pattern

​How it works

​When to use transactions

Use transactions

Skip transactions

​Interaction with retry and DLQ

​Important constraints

​mongoTemplateRef breaks transaction atomicity

​saveEveryN must be 1

​Handler must call saveCheckpointNow()

​Comparison with at-least-once

Reading transaction metadata

Common use cases

Prerequisites for exactly-once

Exactly-once pattern

How it works

When to use transactions

Interaction with retry and DLQ

Important constraints

`mongoTemplateRef` breaks transaction atomicity

`saveEveryN` must be 1

Handler must call `saveCheckpointNow()`

Comparison with at-least-once