Skip to content
Prague
Improve this page

Write a Filter

Goal: decide per message whether it enters the cache, before it costs you an index write, and without a consumer group per tenant.

Prerequisite: a cache registered through AddKafkaCaches (Kafka Integration). Filters are declared on the handler builder, so each block below is a compiled registration method; the one test builds the container with each of them. The code is in WriteAFilter.cs.

1. Header filters: cheapest, run first

Header filters see only the raw headers; neither the key nor the value has been deserialized yet. A string value is compared as UTF-8 with no allocation; WithHeaderFilter<T> decodes the header with MessagePack and hands you the value.

public static void HeaderFilters(KafkaCacheHandlersBuilder b) {
    b.AddCache<OrderCache, string, Order>()
        .WithHeaderEqualsFilter("tenant", "acme")                       // string: UTF-8 compare, no allocation
        .WithHeaderEqualsFilter("region", "EU", "UK")                   // several values: OR within the header
        .WithHeaderNotEqualsFilter("source", "synthetic")
        .WithHeaderExistsFilter("tenant")                               // the header must be present (tombstones exempt)
        .WithHeaderFilter<long>("schema-version", v => v >= 3, passOnNull: false);   // MessagePack-decoded value
}

passOnNull decides what happens when the header is missing (false: reject). Tombstones are exempt from WithHeaderExistsFilter, so a delete without headers still removes the row.

2. Key and value filters

These run after the respective deserialization step, on the single consume thread.

public static void KeyAndValueFilters(KafkaCacheHandlersBuilder b) {
    b.AddCache<OrderCache, string, Order>()
        .WithKeyFilter(static key => !key.StartsWith("TEST-", StringComparison.Ordinal))   // after key deserialization
        .WithValueFilter(static o => o.Total > 0m);                                         // after value deserialization
}

A tombstone (null value) skips key and value filters entirely and is applied as a delete.

3. Turn a soft delete into a removal

By default a rejected message is skipped: the cache keeps whatever it had. With treatAsDelete: true a rejection removes the resident row instead, which is how a Status = Cancelled message becomes a tombstone for the cache.

public static void SoftDeletesBecomeTombstones(KafkaCacheHandlersBuilder b) {
    b.AddCache<OrderCache, string, Order>()
        // A cancelled order is removed from the cache (UpdateType.Delete on the live phase) instead of merely skipped.
        .WithValueFilter(static o => o.Status != OrderStatus.Cancelled, treatAsDelete: true);
}

4. Predicates with DI state

Two overloads. The factory form runs once when the handler is built and passes the result to every call: use it for a snapshot you want frozen (a FrozenSet, a compiled regex). The TService form resolves the service once and passes it to the predicate, so the predicate reads whatever the service holds at the time of the message.

public interface ICustomerAllowList {
    IReadOnlySet<int> CustomerIds { get; }
}

public sealed class StaticCustomerAllowList : ICustomerAllowList {
    public IReadOnlySet<int> CustomerIds { get; } = new HashSet<int> { 42, 7 };
}

public static void FiltersWithDiState(KafkaCacheHandlersBuilder b) {
    b.AddCache<OrderCache, string, Order>()
        // Snapshot: the factory runs once while the handler is built; the predicate gets the frozen set as an argument.
        .WithValueFilter(
            static sp => sp.GetRequiredService<ICustomerAllowList>().CustomerIds.ToFrozenSet(),
            static (allowed, o) => allowed.Contains(o.CustomerId))
        // Live service: TService is spelled out; the predicate reads the service's current state per message.
        .WithValueFilter<ICustomerAllowList>(static (allow, o) => allow.CustomerIds.Contains(o.CustomerId));
}

5. Composition order

Filters AND-compose: a message must pass all of them. The first filter that rejects decides the outcome, so a treatAsDelete filter placed after a plain one never gets to delete what the plain one already skipped.

public static void OrderMatters(KafkaCacheHandlersBuilder b) {
    b.AddCache<OrderCache, string, Order>()
        .WithValueFilter(static o => o.Total > 0m)                                          // rejects → Skip
        .WithValueFilter(static o => o.Status != OrderStatus.Cancelled, treatAsDelete: true); // rejects → Delete
    // A cancelled order with Total 0 is Skipped, not Deleted: the first rejecting filter decides.
}

What a rejection looks like downstream

Phase Skip Delete (treatAsDelete)
Initial load message dropped resident row removed, if any
Live ICacheAfterHandler gets UpdateType.Filtered with no key and no values UpdateType.Delete with the old value, if a row was resident

The decision logic is in Kafka message filters.

Pitfalls

  • Expensive predicates. They run on the consume thread for every message, including the initial replay. Keep them to a dictionary lookup or a comparison; move anything heavier to an after-handler.
  • Filtering on something that changes. If a row entered the cache and a later message for the same key is rejected with the default skip, the stale row stays. Use treatAsDelete when the predicate expresses "should not be here".
  • WithHeaderExistsFilter and tombstones. Tombstones pass it. If you need to drop tombstones without a header, do it with a key filter, which tombstones also bypass; in other words, you cannot filter a tombstone by content, only by header value.
Prague documentation