ChangeStreamContext - FlowWarden

Overview

ChangeStreamContext<T> is the main runtime object passed to every handler method (@OnChange, @OnInsert, @OnUpdate, @OnDelete, @OnReplace). It provides access to the event metadata, documents, and runtime actions.

Document access

getFullDocument

Returns the full document after the operation, automatically converted to the requested type using Spring’s MongoConverter.

@OnChange
void handle(ChangeStreamContext<Order> ctx) {
    Optional<Order> order = ctx.getFullDocument(Order.class);
}

For DELETE operations, the full document is not available and the method returns Optional.empty().

getFullDocumentBeforeChange

Returns the document before the operation, automatically converted using Spring’s MongoConverter. This is useful for comparing old vs. new state (e.g., audit logging, status transition tracking).

@OnUpdate
void onStatusChange(ChangeStreamContext<Order> ctx) {
    Optional<Order> before = ctx.getFullDocumentBeforeChange(Order.class);
    Optional<Order> after = ctx.getFullDocument(Order.class);

    if (before.isPresent() && after.isPresent()) {
        log.info("Order {} status: {} → {}",
            after.get().getId(),
            before.get().getStatus(),
            after.get().getStatus());
    }
}

Two prerequisites are required for pre-images to work:

MongoDB 6.0+ — pre-images are not available on earlier versions.
Pre-images must be enabled on the collection — MongoDB does not store pre-images by default. Run this command once per collection:

db.runCommand({
  collMod: "orders",
  changeStreamPreAndPostImages: { enabled: true }
})

Or in Java with Spring’s MongoTemplate:

mongoTemplate.getDb().runCommand(
    new Document("collMod", "orders")
        .append("changeStreamPreAndPostImages", new Document("enabled", true))
);

The @ChangeStream annotation controls how pre-images are requested:

`fullDocumentBeforeChange`	Behavior if pre-images are not enabled on collection
`OFF` (default)	No pre-image requested. `getFullDocumentBeforeChange()` always returns `Optional.empty()`.
`WHEN_AVAILABLE`	Returns `Optional.empty()` silently. Safe for optional use cases.
`REQUIRED`	MongoDB returns an error and the event processing fails. Use when pre-images are mandatory.

@ChangeStream(
    documentType = Order.class,
    fullDocumentBeforeChange = FullDocumentBeforeChangeMode.WHEN_AVAILABLE
)
public class OrderAuditHandler { ... }

getUpdateDescription

For UPDATE operations, returns an UpdateDescription describing the modified fields. Returns Optional.empty() for non-UPDATE operations.

ctx.getUpdateDescription().ifPresent(desc -> {
    Map<String, Object> updated = desc.getUpdatedFields();
    List<String> removed = desc.getRemovedFields();
});

UpdateDescription methods

Method	Returns	Description
`getUpdatedFields()`	`Map<String, Object>`	All fields that were updated and their new values
`getRemovedFields()`	`List<String>`	Dot-notation paths of fields that were removed
`getTruncatedArrays()`	`List<TruncatedArray>`	Arrays that were truncated during the update (e.g. `$pop` operations)
`hasFieldChanged(String path)`	`boolean`	Whether the given field was modified (updated or removed)
`getUpdatedFieldValue(String path, Class<V> type)`	`Optional<V>`	The new value of a specific updated field, typed

Targeted field changes

hasFieldChanged and getUpdatedFieldValue let you react to specific fields without manually walking the updated-fields map:

@OnUpdate
void onOrderUpdate(ChangeStreamContext<Order> ctx) {
    ctx.getUpdateDescription().ifPresent(update -> {
        if (update.hasFieldChanged("status")) {
            String newStatus = update.getUpdatedFieldValue("status", String.class)
                .orElseThrow();
            log.info("Order {} status changed to {}", ctx.getDocumentKey(), newStatus);
        }
    });
}

Paths use dot notation for nested fields (e.g. "address.city").

Custom conversion — By default, getFullDocument(MyType.class) uses Spring’s MongoConverter, which respects your @Field annotations, custom converters registered in MongoCustomConversions, etc.If you need a completely different conversion strategy (e.g. Jackson, Gson), retrieve the raw BSON document and convert it yourself:

@OnChange
void handle(ChangeStreamContext<?> ctx) {
    Document raw = ctx.getFullDocument(Document.class).orElse(null);
    if (raw != null) {
        MyType obj = myCustomMapper.fromBson(raw.toJson());
    }
}

Event metadata

Method	Returns	Description
`getEventId()`	`String`	Unique ID generated by FlowWarden for this event
`getOperationType()`	`OperationType`	`INSERT`, `UPDATE`, `REPLACE`, `DELETE`, `DROP`, `INVALIDATE`
`getStreamName()`	`String`	Name of the `@ChangeStream` handler
`getCollectionName()`	`String`	Source MongoDB collection
`getDatabaseName()`	`String`	Source MongoDB database
`getClusterTime()`	`Instant`	MongoDB cluster timestamp (second precision)
`getWallTime()`	`Instant`	MongoDB wall-clock timestamp (millisecond precision, MongoDB 6.0+). Returns `null` if unavailable.
`getResumeToken()`	`BsonDocument`	Resume token for checkpointing
`getTransactionInfo()`	`Optional<TransactionInfo>`	MongoDB transaction metadata (`lsid`, `txnNumber`) when the event is part of a transaction; `Optional.empty()` for standalone operations
`getDocumentKey()`	`BsonValue`	The `_id` of the affected document
`getAttemptNumber()`	`int`	Current retry attempt (1-based)
`summary()`	`String`	Human-readable summary of the event, suitable for logging

TransactionInfo

public record TransactionInfo(BsonDocument lsid, long txnNumber) {}

MongoDB emits lsid (logical session id) and txnNumber together for every event in a transaction, or omits both for standalone operations. They’re grouped in a record so handlers cannot observe one without the other; absence is represented by Optional.empty() at the call site. Useful for grouping events of the same transaction — audit logs, aggregation, atomicity-preserving downstream propagation.

Actions

sendToDlq

Manually routes the current event to the Dead Letter Queue.

ctx.sendToDlq("Invalid order total");

saveCheckpointNow

Forces an immediate checkpoint save for the current resume token.

ctx.saveCheckpointNow();

Custom metadata

Attach arbitrary key-value pairs to the event for downstream processing.

ctx.addMetadata("processingRegion", "eu-west-1");

Optional<String> region = ctx.getMetadata("processingRegion", String.class);
Map<String, Object> all = ctx.getAllMetadata(); // unmodifiable

​Overview

​Document access

​getFullDocument

​getFullDocumentBeforeChange

​getUpdateDescription

​UpdateDescription methods

​Targeted field changes

​Event metadata

​TransactionInfo

​Actions

​sendToDlq

​saveCheckpointNow

​Custom metadata