Filters
Let the server send each consumer group only the messages it needs
A filter belongs to a consumer group. The server runs it next to the data and sends the group only the messages that match. Each delivered message keeps its original offset, headers, and payload bytes. Use filters when one shared topic feeds many consumers that each care about a small part of it, such as alert routing, change data capture (CDC), or backfills that skip most of a topic.
In the CDC example, the reader receives 4 of 240 records and 424 of 27,953 payload bytes, so 98.5% of the payload never leaves the server. Rust, Python, and TypeScript produce the same result.
Quick example
A monitoring agent wants only the readings where CPU is at 90 or above. Create the group with its filter once. Every consumer of that group then gets only the matches.
import { ConsumerFilter, FilterExpr, Laser } from "@laserdata/laser-sdk"
await using laser = await Laser.connectEnv()
const topic = laser.stream("telemetry").topic("metrics")
await topic.ensure(2)
for (const [host, cpu] of [["node-1", 42], ["node-2", 97], ["node-3", 55]] as const) {
await topic.publish().json({ host, cpu }).send()
}
// One-time setup. The group owns the filter.
const hot = topic.consumerGroup("hot-hosts")
await hot.create({ filter: ConsumerFilter.json(FilterExpr.pred("cpu", "gte", 90)) })
await using consumer = await hot.consumer({
startAt: { kind: "first" },
commitPolicy: { kind: "disabled" }
})
for await (const message of consumer) {
console.log(message.position.offset, message.json())
await consumer.commit(message)
}use laser_sdk::filters::{ConsumerFilter, FilterExpr};
use laser_sdk::prelude::*;
use laser_sdk::query::CmpOp;
use serde::{Deserialize, Serialize};
#[derive(Debug, Serialize, Deserialize)]
struct Reading {
host: String,
cpu: u32,
}
#[tokio::main]
async fn main() -> Result<(), LaserError> {
let laser = Laser::connect_env().await?;
let topic = laser.stream("telemetry").topic("metrics");
topic.ensure(2).await?;
for (host, cpu) in [("node-1", 42), ("node-2", 97), ("node-3", 55)] {
let reading = Reading { host: host.into(), cpu };
topic.publish().json(&reading)?.send().await?;
}
// One-time setup. The group owns the filter.
let hot = topic.consumer_group("hot-hosts");
hot.create()
.filter(ConsumerFilter::json(FilterExpr::pred("cpu", CmpOp::Gte, 90_i64)))
.build()
.await?;
let mut consumer = hot
.consumer()
.start_at(ConsumerStart::First)
.commit_policy(CommitPolicy::Disabled)
.build()
.await?;
while let Some(message) = consumer.next().await {
let message = message?;
let reading: Reading = message.json()?;
println!("{} {reading:?}", message.position.offset);
consumer.commit(&message).await?;
}
consumer.shutdown().await
}import asyncio
import laser_sdk as ls
async def main() -> None:
laser = await ls.Laser.connect_env()
topic = laser.stream("telemetry").topic("metrics")
await topic.ensure(2)
for host, cpu in (("node-1", 42), ("node-2", 97), ("node-3", 55)):
await topic.publish().json({"host": host, "cpu": cpu}).send()
# One-time setup. The group owns the filter.
hot = topic.consumer_group("hot-hosts")
await hot.create(filter=ls.ConsumerFilter.json(ls.FilterExpr.pred("cpu", "gte", 90)))
consumer = hot.consumer(polling="first", auto_commit="disabled")
try:
async for message in consumer:
print(message.position.offset, message.json())
await consumer.commit(message)
finally:
await consumer.shutdown()
asyncio.run(main())Only the node-2 reading arrives. The consumer names the group, not the filter. Run create from a setup step. It is safe to repeat with the same filter.
Combine conditions
Build an expression from predicates, then wrap it in the codec of the payload. all needs every child to match, any needs one, and negate inverts a child. Field paths use dots for object keys and [n] for array items, such as after.ground_stations[0].
const critical = ConsumerFilter.json(
FilterExpr.all([
FilterExpr.pred("severity", "eq", "critical"),
FilterExpr.any([
FilterExpr.pred("service", "eq", "billing-agent"),
FilterExpr.pred("service", "eq", "triage-agent")
])
])
)let critical = ConsumerFilter::json(FilterExpr::all([
FilterExpr::pred("severity", CmpOp::Eq, "critical"),
FilterExpr::any([
FilterExpr::pred("service", CmpOp::Eq, "billing-agent"),
FilterExpr::pred("service", CmpOp::Eq, "triage-agent"),
]),
]));critical = ls.ConsumerFilter.json(
ls.FilterExpr.all([
ls.FilterExpr.pred("severity", "eq", "critical"),
ls.FilterExpr.any([
ls.FilterExpr.pred("service", "eq", "billing-agent"),
ls.FilterExpr.pred("service", "eq", "triage-agent"),
]),
])
)Comparisons are eq, ne, lt, lte, gt, gte, in, contains, and prefix. Text matching, presence checks, and timestamp coercion are in Filters in depth. JSON, CBOR, Avro, and Protobuf payloads all work.
Filter on headers without decoding
A headers_only filter reads typed message headers and never decodes the payload, so it works with any payload format and costs the least to run.
const pager = laser.stream("ops").topic("alerts").consumerGroup("pager")
await pager.create({
filter: ConsumerFilter.headersOnly(FilterExpr.header("priority", "eq", 2))
})let pager = laser.stream("ops").topic("alerts").consumer_group("pager");
pager
.create()
.filter(ConsumerFilter::headers_only(FilterExpr::header("priority", CmpOp::Eq, 2_i32)))
.build()
.await?;pager = laser.stream("ops").topic("alerts").consumer_group("pager")
await pager.create(
filter=ls.ConsumerFilter.headers_only(ls.FilterExpr.header("priority", "eq", 2))
)Producers set typed headers through topic.producer(). See Filter on headers only.
Check a filter before you rely on it
test judges one sample payload and explains each predicate. preview judges stored messages in one partition. Neither joins the group or stores progress.
const tested = await hot.filter().test(JSON.stringify({ host: "node-9", cpu: 93 }))
console.log(tested.explanation.verdict)
const preview = await hot.filter().preview(0, { maxRecords: 10 })
console.log(preview.examined, preview.matched)let sample = r#"{"host":"node-9","cpu":93}"#;
let tested = hot.filter().test(sample, Vec::new()).await?;
println!("{:?}", tested.explanation.verdict);
let preview = hot.filter().preview(0).await?.max_records(10).send().await?;
println!("{} {}", preview.examined, preview.matched);tested = await hot.filter().test('{"host":"node-9","cpu":93}')
print(tested["explanation"]["verdict"])
preview = await hot.filter().preview(0, max_records=10)
print(preview["examined"], preview["matched"])Change or remove a filter
A group keeps the filter it was created with. Configuring a different filter on the same group fails with conflict, so two selections never share one set of offsets. Create a new group for a new selection. group.filter().release() removes the group's filter, and its consumers then receive every message. A released group, or one whose filter was deleted, can take a filter again only with the same digest it ran before.
Good to know
- Configuring a filter needs
laser-plane, so use Laser Stack or LaserData Cloud. Plain Apache Iggy has no filters. - A group with no filter receives every message. A read never falls back to unfiltered delivery when the filter cannot run. It fails instead.
- A batch of 100 means the server examines up to 100 messages per partition. It can deliver fewer, or none, and an empty poll does not mean the topic is empty.
- Progress covers the messages the server skipped, so a group that matches one message in a million still moves forward.
- Raw Apache Iggy consumers ignore the filter and receive every message. Do not mix them with Laser SDK consumers on one group. A filter is not an access control boundary.
- Filters judge each message on its own. They do not compare it with an earlier message.