> ## Documentation Index
> Fetch the complete documentation index at: https://docs.flowwarden.io/llms.txt
> Use this file to discover all available pages before exploring further.

# @Filter

> Application-side Java filtering for Change Stream events

## 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`](/reference/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.

<Note>
  At most **one** `@Filter` method is allowed per `@ChangeStream` class.
</Note>

## Supported Signatures

<CodeGroup>
  ```java Predicate style theme={null}
  @Filter
  Predicate<ChangeStreamContext<?>> myFilter() {
      return ctx -> "PAID".equals(
          ctx.getFullDocument(Order.class).getStatus()
      );
  }
  ```

  ```java Boolean method style theme={null}
  @Filter
  boolean filter(ChangeStreamContext<?> ctx) {
      return "PAID".equals(
          ctx.getFullDocument(Order.class).getStatus()
      );
  }
  ```
</CodeGroup>

| 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

```java theme={null}
@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**:

<Warning>
  This configuration causes a `BeanCreationException` at startup — the application will not start.
</Warning>

**`@Filter` + typed handler for a no-fullDocument operation**

```java theme={null}
// 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**:

```java theme={null}
// 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; }
}
```

```java theme={null}
// 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**                                                               |

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

## 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](/reference/checkpoint#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`](/reference/pipeline) on the same `@ChangeStream`,
forming a **double filtering funnel**:

```mermaid theme={null}
flowchart TD
    A[("Oplog (all writes)")] -->|"implicit collection filter"| B["Change Stream on collection"]
    B -->|"server-side"| C["@Pipeline<br/>MongoDB aggregation"]
    C -->|"only matching events<br/>sent over the wire"| D{"@Filter<br/>Java predicate"}
    D -->|"false"| X(["discarded"])
    D -->|"true"| E["Handler invoked"]
```

**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.

```java theme={null}
@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
    }
}
```

<Note>
  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](/reference/checkpoint#resume-cascade)
  level-2 fallback on streams where `lastProcessedToken` lags behind.
</Note>

## See Also

<CardGroup cols={2}>
  <Card title="@Pipeline" icon="server" href="/reference/pipeline">
    Server-side aggregation pipeline filtering
  </Card>

  <Card title="Filtering Events Guide" icon="book" href="/guides/filtering-events">
    Complete guide combining @Pipeline and @Filter
  </Card>

  <Card title="@Checkpoint" icon="floppy-disk" href="/reference/checkpoint">
    Resume token persistence and dual checkpoint
  </Card>

  <Card title="Event Handlers" icon="bolt" href="/reference/event-handlers">
    @OnInsert, @OnUpdate, @OnDelete, @OnChange
  </Card>
</CardGroup>