Skip to main content

​
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 — but their resume token is still checkpointed as lastSeenToken. 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 typeParametersDescription
Predicate<ChangeStreamContext<?>>NoneReturns a reusable predicate. Called once, predicate tested per event.
booleanChangeStreamContext<?>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?
@OnInsertYes
@OnUpdateYes
@OnReplaceYes
@OnInsert + @OnUpdateYes
@OnDeleteNo
@OnChangeYes (predicate must handle Optional.empty() for DROP / INVALIDATE)
@OnInsert + @OnDeleteNo
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

Events rejected by @Filter are not forwarded to the handler, so they never advance lastProcessedToken. Their resume token does, however, advance the in-memory lastSeenToken tracker, which the heartbeat timer (@Checkpoint(saveIntervalSeconds)) persists on each tick. This is what enables the resume cascade level-2 fallback on streams where most events are filter-rejected. On filter-heavy streams, a gap grows between lastSeenToken (advancing) and lastProcessedToken (stalled until a non-filtered event succeeds). Keep saveIntervalSeconds > 0 so the heartbeat can persist lastSeenToken regularly — otherwise the cascade has no level-2 safety net.

​
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 advance lastSeenToken regardless of whether @Filter accepts them. The heartbeat timer persists lastSeenToken, enabling the resume cascade level-2 fallback on streams where lastProcessedToken lags behind.

​
See Also

@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