Consumer Filters: Stop Moving Data You Do Not Need
LaserData consumer filters select records beside the log, cutting network transfer and downstream work while preserving replay, offsets, and flexible consumer policies.
Most data platforms spend too much money moving data that applications immediately discard.
A change data capture feed might carry every update from every table, while a fraud service needs only high-value order changes. An operations desk might care about four event types in a topic containing hundreds. A backfill might need 2% of a year-long stream. In the usual architecture, each consumer downloads the full feed, deserializes every record, applies its own conditions, and throws almost everything away.
That waste compounds as the business grows. More consumers mean more network transfer, more application CPU, more memory pressure, more autoscaling, and often more derived topics created only to route subsets of the same data.
Consumer filters in the Laser SDK change where that work happens. The consumer declares what it needs. LaserData evaluates the policy beside the durable log and returns only matching records across the network.
The idea is simple. Its effect on the cost, latency, and design of a streaming platform is not.
Open full size ↗The executive case: make data movement proportional to business value
Streaming costs are not limited to storage. Every record a consumer does not need can still create costs across the system:
The hidden cost of client-side rejection
The record is discarded last. The infrastructure bill starts first.
Network
- Server egress
- Cross-zone traffic
Application
- Consumer compute
- Decoding
- Memory
Reliability & operations
- Retries
- Operations
Cumulative delivery cost
Paid before the record is rejected
Filter earlier, and the non-match never reaches the network or application stages.
Client-side filtering pays most of that bill before deciding that a record is irrelevant. Server-side filtering moves the decision to the point where the data already lives.
The complete Rust CDC example makes the difference concrete. It publishes a mixed satellite operations feed containing battery readings, maneuvers, ground-station changes, safe-mode events, and decommissions. The anomaly desk needs only safe-mode and decommission events.
The result:
| Measure | Full feed | Filtered delivery |
|---|---|---|
| Records | 240 | 4 |
| Payload bytes | 27,953 | 424 |
| Payload transfer avoided | 0% | 98.5% |
Open full size ↗This is a deliberately small, reproducible example, not a projection from a spreadsheet. The Frostline example runs a larger CDC workload: ten million change records, 6.30 GB of payload, and four filtered consumer groups. In its recorded run, the groups received 777.64 MB instead of the 25.21 GB that four full-feed reads would move, avoiding 96.9% of payload transfer. Counting both TCP directions at the socket, application traffic fell from 31.40 GB to 1.11 GB, a 96.5% reduction. That run predates the group-owned filter API of Laser SDK 0.5.1, and the repository publishes the whole benchmark, including server CPU and latency.
The exact savings depend on how selective each policy is. The architectural implication is broader: network and downstream processing can scale with the records that create value for a consumer, rather than with the total volume retained in the source stream.
For a CTO or VP of Engineering, that creates four forms of leverage.
Lower unit cost without weakening the event backbone
Teams can keep a complete, replayable source of truth while reducing the bytes delivered to selective consumers. They do not have to trade retention and optionality for a smaller application footprint.
Fewer routing pipelines to own
Without server-side filters, organizations often add a stream processor or create intermediate topics for every audience: fraud events, premium-customer events, compliance events, operational alerts, and regional subsets. Each new branch adds configuration, capacity planning, monitoring, access rules, retention, and failure recovery.
Consumer filters let multiple groups follow different subsets of one durable topic. The log remains shared. Delivery policy varies by consumer group.
Better economics for cloud and hybrid architectures
Filtering before transfer matters when a producer and consumer sit in different availability zones, regions, clouds, or trust zones. It also matters for branch locations and edge environments where bandwidth is finite. Avoided bytes reduce both infrastructure load and the amount of data crossing expensive boundaries.
Faster delivery of new use cases
A new consumer can declare a policy against an existing feed instead of waiting for a new routing service and topic hierarchy. That shortens the path from a business question to a production data product. It also reduces the number of permanent components created for temporary experiments.
The architecture: one log, many selective views of motion
A filtered reader asks for a page from a topic partition. The server scans source records, evaluates the consumer's declarative filter, and returns the matches. Nonmatching records stay on the server.
Matching records are not rewritten. They keep their original offset, message ID, timestamp, headers, and payload bytes. Applications can continue to use their existing typed decoders, audit identifiers, and replay logic.
This distinction is important. A consumer filter is not a materialized copy and does not create a second stream of transformed records. It is a delivery policy over the durable log.
Each group maintains its own progress. One group can be selective without changing what another group reads or what the topic retains.
What happens to latency
Moving less data is usually good for end-to-end efficiency, but it is useful to be precise about where the gains come from.
A filtered consumer can avoid:
- transferring rejected payloads
- allocating memory for them in the application
- decoding and validating them downstream
- running application-level rejection logic
- passing them through additional internal pipelines or services
That can reduce consumer CPU time, network wait, garbage collection in managed runtimes, and the time between receiving a relevant record and acting on it. It can also reduce contention when many selective consumers share the same source.
Filtering is not free. The streaming server still examines records for each reader. Payload predicates require decoding work on the server, and a very sparse filter may scan many source records before finding enough matches to fill a page. LaserData therefore keeps match limits and scan budgets explicit.
The Frostline run shows the trade. Reader CPU fell from 66.78 to 5.91 seconds, while streaming server CPU rose from 31.32 to 99.50 seconds, so combined CPU rose by about 9.5%. On local loopback, where network transfer costs almost nothing, the filtered read phase took 29.97 seconds against 23.38 seconds for ordinary readers. Filtering pays off when network bytes, consumer compute, or downstream work cost more than the extra scanning on the server.
The least expensive route is often a typed header. A header-only predicate never parses the payload. On a JSON payload, parsing costs more than the comparison itself, so a header test costs much less than a field or text predicate. The SDK ships filter_evaluation and filter_paths benchmarks to measure this on your own records. Record size, codec, expression shape, selectivity, storage, and deployment topology all matter.
A practical design rule is:
- Put a stable routing field such as
event.type, tenant, region, or priority in a typed header. - Reject unrelated records with a header predicate first.
- Add payload predicates only when the header match is not specific enough.
- Benchmark with the real record distribution, including p99 and p99.9 behavior.
In an all expression, header conditions run first and evaluation stops at the first non-match. That lets a mixed-format topic reject most unrelated records without decoding their payloads.
A filter is a policy, not application code on the server
LaserData filters are declarative. Consumers do not upload executable code into the data plane. They compose bounded expressions from a defined set of operations:
- comparisons such as
eq,ne,lt,lte,gt,gte,in,contains, andprefix - text operations including equals, prefix, suffix, contains, glob, and regex
- typed-header comparisons
- presence and absence tests
all,any, andnotcomposition- explicit coercion for values such as timestamp and decimal strings
Payload filtering supports JSON, CBOR, Avro, and Protobuf. Avro and Protobuf use registered writer schemas. A headers_only filter works with any payload format because it never decodes the body.
This gives architects room to use the same mechanism across CDC, event routing, backfills, alerting, and mixed-format integration topics without embedding a general-purpose runtime inside the streaming server.
From a CDC requirement to a group policy in Rust, Python, and TypeScript
Suppose a commerce change feed contains every order update, but a risk service needs only orders whose new total is at least 10,000. The same filter can be attached when the consumer group is created in Rust, Python, or TypeScript:
use laser_sdk::filters::{ConsumerFilter, FilterExpr, FilteredStart};use laser_sdk::prelude::full::*;use laser_sdk::query::CmpOp; let risk_policy = ConsumerFilter::json(FilterExpr::all([ FilterExpr::pred("table", CmpOp::Eq, "orders"), FilterExpr::pred("op", CmpOp::Eq, "u"), FilterExpr::pred("after.total", CmpOp::Gte, 10_000_i64),])); let group = laser .stream("commerce") .topic("commerce_cdc") .consumer_group("risk-review"); group.create().filter(risk_policy).build().await?; let mut reader = group .reader()? .start(FilteredStart::First) .build() .await?;All three build the same definition with the same digest, so the idempotent setup can run from any of them. After setup, members can join by group name or ID. They do not need to repeat the filter definition. The group owns the delivery policy, and native partition assignment still distributes work among members.
That separation is useful for platform teams. Application code expresses how to process a qualifying event. The group binding defines which events qualify.
The CDC example shows more than byte savings
The Rust example is worth reading because it exercises the operational lifecycle, not only the happy path.
Evidence of change, not only current value
The transition branch of the satellite policy does not merely ask whether after.mode equals safe. It also requires the producer-supplied changed list to contain mode. A battery update for a satellite already in safe mode therefore does not look like a new transition. The policy's other branches select decommissions and telemetry reports of safe mode, which carry no such evidence.
This is a critical CDC design choice. A filter evaluates one record at a time. It does not infer a prior row version by searching the log. If a consumer needs evidence that a specific column changed, the producer or CDC envelope should carry that evidence.
Header-only routing for opaque events
The example publishes binary alerts with a numeric priority header. A pager group selects critical alerts without decoding their bodies. That pattern works for encrypted, proprietary, or mixed-codec payloads as long as producers stamp trustworthy routing metadata.
The same policy model across codecs
The example publishes typed records in CBOR, Avro, and Protobuf, then applies a field predicate to each. The consumer receives the original encoded bytes and decodes them using its normal schema-aware path.
Test and preview before consumers read
Once a group holds its policy, teams can test that policy against a supplied sample record or preview it against a bounded range of stored records in one partition. Neither operation joins the group or advances an offset, so verdicts and match rates can be inspected before any consumer starts reading. Before a group exists, the SDK's local evaluator gives the same verdict for a sample record without a server.
Controlled revisions instead of silent policy drift
Saved filters have immutable revisions. The example drafts a stricter revision while the group keeps running its active one, then runs the stricter definition in a separate A/B group. It pauses the variant's revision, acknowledges in-flight work, resumes it, and shows that a group running one policy refuses another with conflict instead of switching predicates over the same offsets.
This is more than configuration hygiene. If two different filters shared progress, one version could advance past records required by the other. Separate groups and immutable revisions keep delivery semantics explicit and reviewable.
Open full size ↗Safe progress across records that never arrive
Filtering creates a subtle offset problem: how does a consumer make progress across source records it never receives?
LaserData tracks both delivered work and the range examined by the server. Empty pages can advance their scanned range automatically. For pages with matches, the reader lets application work finish and stores only the offset through which every returned record is complete. Applications can process returned records out of order without committing past unfinished work.
The reader also bounds unacknowledged record-bearing pages per partition. If that bound is reached, it stops reading the partition until work is acknowledged rather than allowing read-ahead memory to grow without limit.
These mechanics preserve the properties enterprises depend on from consumer groups: resumability, replay, partition ownership, bounded memory, and deliberate acknowledgment.
Design tradeoffs architects should make explicit
Consumer filters remove an entire class of avoidable data movement, but they should be treated as an engineered policy layer.
Selectivity determines the return
A consumer that needs 95% of a topic will save far less than one that needs 1%. Measure payload bytes, not only record counts. A small number of large records can dominate transfer.
Put cheap predicates first
Prefer typed headers for coarse routing and payload fields for precise decisions. This is especially important for topics that mix codecs or event families.
Define malformed-record behavior
A payload that cannot be decoded is a fault, not a normal non-match. The filter can stop, pass the record through as unevaluated, or drop it. The right choice depends on whether continuity, quarantine, or strict correctness is the priority.
Keep authorization separate
A consumer filter is a delivery policy, not an access-control boundary. Use topic permissions and deployment authorization to control which principals may read data at all.
Treat policy changes like schema changes
Version them, preview them, test representative samples, run variants in separate groups, observe their match rates and latency, and roll them out deliberately.
Where consumer filters fit best
The strongest candidates are workloads where one durable feed serves multiple narrow interests:
- CDC feeds shared by search indexing, fraud, analytics, cache invalidation, and audit consumers
- operational alert streams routed by severity, service, tenant, or region
- large backfills where a repair job needs only a small historical subset
- agent and automation systems that react to specific tool results, state transitions, or exceptions
- multi-tenant event platforms where services consume only their assigned domain
- edge or cross-region consumers with constrained or expensive bandwidth
The common pattern is high source volume, low matching fraction, and several consumers with different definitions of relevance.
The larger platform decision
For years, streaming platforms have made it normal to move a complete dataset into every application and ask each team to reduce it locally. That approach is simple at the beginning. At enterprise scale, it turns relevance into repeated infrastructure work.
Consumer filters make a different trade: keep one complete, durable log, but move only the records each consumer can act on.
For engineering leaders, that means lower transfer costs, less consumer compute, fewer derived pipelines, and a platform that can support more use cases without multiplying infrastructure. For lead engineers and architects, it means declarative server-side selection with original offsets, typed payloads, consumer-group progress, testable policies, and controlled revisions.
The goal is not to make the streaming server do arbitrary application work. It is to stop every application from paying to receive data it already knows it does not need.
Read the consumer filters guide →