@Filter

Overview

@Filter performs application-side filtering in Java. All events arrive from MongoDB, and a predicate decides which ones are forwarded to handler methods. Events that don’t pass the filter are silently skipped — a rejection is a terminal settlement, so their resume token still advances the lastProcessedToken checkpoint anchor. Unlike @Pipeline which runs a MongoDB aggregation pipeline once at stream startup to reduce network traffic, @Filter executes on every event and can leverage Spring beans, service calls, and any Java logic.

At most one @Filter method is allowed per @ChangeStream class.

Supported Signatures

@Filter
Predicate<ChangeStreamContext<?>> myFilter() {
    return ctx -> "PAID".equals(
        ctx.getFullDocument(Order.class).getStatus()
    );
}

@Filter
boolean filter(ChangeStreamContext<?> ctx) {
    return "PAID".equals(
        ctx.getFullDocument(Order.class).getStatus()
    );
}

Return type	Parameters	Description
`Predicate<ChangeStreamContext<?>>`	None	Returns a reusable predicate. Called once, predicate tested per event.
`boolean`	`ChangeStreamContext<?>`	Direct evaluation. Called on every event.

Basic Example

@ChangeStream(collection = "orders", documentType = Order.class)
public class PaidOrderHandler {

    @Filter
    boolean onlyPaid(ChangeStreamContext<?> ctx) {
        Order order = ctx.getFullDocument(Order.class);
        return order != null && "PAID".equals(order.getStatus());
    }

    @OnInsert
    void handle(Order order, ChangeStreamContext<Order> ctx) {
        // Only receives events where status == PAID
        log.info("Paid order: {}", order.getId());
    }
}

Startup Validation (Fail-Fast)

FlowWarden validates @Filter compatibility at application startup and rejects clearly incompatible configurations immediately with a clear error message. The application will fail to start if @Filter is combined with a typed handler covering an operation where MongoDB does not provide a fullDocument.

Why?

@Filter predicates typically call ctx.getFullDocument(), which returns Optional.empty() for DELETE, DROP, and INVALIDATE operations. Running the filter on a typed handler dedicated to one of those operations is almost always a mistake, so FlowWarden rejects that combination at startup. Combining @Filter with the @OnChange catch-all is allowed — DROP and INVALIDATE will reach the predicate, and the predicate must handle Optional.empty() explicitly.

Rule

The following combination is rejected:

This configuration causes a BeanCreationException at startup — the application will not start.

@Filter + typed handler for a no-fullDocument operation

// REJECTED: @OnDelete handles DELETE which has no fullDocument
@ChangeStream(collection = "orders")
public class InvalidHandler {

    @OnDelete
    void onDelete(ChangeStreamContext<?> ctx) { }

    @Filter
    boolean filter(ChangeStreamContext<?> ctx) { return true; }
}

The error message:

@ChangeStream class InvalidHandler declares @Filter and @OnDelete, which is not allowed.
DELETE events have no fullDocument, so the @Filter predicate cannot safely access the document.
Use a server-side @Pipeline to filter these events, or move the filtering logic into the
handler method.

Valid Combinations

The following combinations are accepted:

// OK: @Filter + typed handlers for operations WITH fullDocument
@ChangeStream(collection = "orders")
public class ValidHandler {

    @OnInsert
    void onInsert(ChangeStreamContext<?> ctx) { }

    @OnUpdate
    void onUpdate(ChangeStreamContext<?> ctx) { }

    @Filter
    boolean filter(ChangeStreamContext<?> ctx) { return true; }
}

// OK: @Filter + @OnChange catch-all — the predicate guards against missing fullDocument
@ChangeStream(collection = "orders", documentType = Order.class)
public class ValidHandler2 {

    @OnChange
    void handle(ChangeStreamContext<Order> ctx) { }

    @Filter
    boolean filter(ChangeStreamContext<Order> ctx) {
        return ctx.getFullDocument(Order.class)
                  .map(o -> "PAID".equals(o.getStatus()))
                  .orElse(false);  // DROP / INVALIDATE / DELETE → no fullDocument → drop
    }
}

Quick Reference

Handler combination	`@Filter` allowed?
`@OnInsert`	Yes
`@OnUpdate`	Yes
`@OnReplace`	Yes
`@OnInsert` + `@OnUpdate`	Yes
`@OnDelete`	No
`@OnChange`	Yes (predicate must handle `Optional.empty()` for DROP / INVALIDATE)
`@OnInsert` + `@OnDelete`	No

If you need to handle DELETE events alongside other operations while filtering, use @Pipeline for server-side filtering, guard the @Filter predicate against Optional.empty(), or move the filter logic directly into your handler method.

Checkpoint Interaction

A rejection by @Filter is a terminal settlement: the event’s outcome is decided, so its token advances lastProcessedToken exactly like a handler success does (per the saveEveryN / saveIntervalSeconds policy). A filter-heavy stream therefore stays recoverable at resume cascade level 1 through its rejected traffic: rejections advance the anchor instead of stalling it. As with any settlement, a restart re-reads at most the events since the last confirmed write (up to N-1 with saveEveryN = N) — those rejections are simply re-evaluated by the filter. Rejected events never touch lastSeenToken, which holds exclusively server-certified positions written by the idle heartbeat (idleHeartbeatIntervalSeconds) — the safety net for periods where the stream settles nothing at all.

Combining with @Pipeline

@Filter can coexist with @Pipeline on the same @ChangeStream, forming a double filtering funnel: Typical use case: pre-filter operationType = insert | update and status = PAID server-side with @Pipeline, then verify application-side with @Filter that the tenant is active via a Spring service call.

@ChangeStream(collection = "orders", documentType = Order.class)
public class PaidOrderHandler {

    @Pipeline
    List<Bson> pipeline() {
        return List.of(
            Aggregates.match(Filters.and(
                Filters.in("operationType", "insert", "update"),
                Filters.eq("fullDocument.status", "PAID")
            ))
        );
    }

    @Filter
    boolean filter(ChangeStreamContext<?> ctx) {
        // @Pipeline already filtered server-side; refine with Java logic
        return tenantService.isActive(ctx.getFullDocument(Order.class).getTenantId());
    }

    @OnInsert
    void onNewPaidOrder(Order order, ChangeStreamContext<Order> ctx) {
        // Only receives PAID orders from active tenants
    }
}

When combining @Pipeline + @Filter, events that pass the server-side pipeline reach FlowWarden and settle terminally regardless of whether @Filter accepts them — accepted or rejected, they advance the lastProcessedToken anchor. When the pipeline filters out everything for a while, the idle heartbeat takes over and certifies fresh positions into lastSeenToken, preserving the resume cascade level-2 safety net.

@Pipeline

Server-side aggregation pipeline filtering

Filtering Events Guide

Complete guide combining @Pipeline and @Filter

@Checkpoint

Resume token persistence and dual checkpoint

Event Handlers

@OnInsert, @OnUpdate, @OnDelete, @OnChange

Annotations

Registration API

Runtime

SPI

Overview

Supported Signatures

Basic Example

Startup Validation (Fail-Fast)

Why?

Rule

Valid Combinations

Quick Reference

Checkpoint Interaction

Combining with @Pipeline

See Also

@Pipeline

Filtering Events Guide

@Checkpoint

Event Handlers

​Overview

​Supported Signatures

​Basic Example

​Startup Validation (Fail-Fast)

​Why?

​Rule

​Valid Combinations

​Quick Reference

​Checkpoint Interaction

​Combining with @Pipeline

​See Also

@Pipeline

Filtering Events Guide

@Checkpoint

Event Handlers

Overview

Supported Signatures

Basic Example

Startup Validation (Fail-Fast)

Why?

Rule

Valid Combinations

Quick Reference

Checkpoint Interaction

Combining with @Pipeline

See Also