From f54cbccff387cb7ea6bb1debbef2ea05502035dc Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Wed, 8 Jul 2026 21:05:54 -0700 Subject: [PATCH] [adapters] Kafka header filters. New Kafka input connector config setting to filter messages based on Kafka message headers. Supports primitive checks against regular expressions and arbitrary Boolean combinations. Signed-off-by: Leonid Ryzhyk --- Cargo.lock | 1 + .../adapters/src/transport/kafka/ft/input.rs | 56 ++- .../adapters/src/transport/kafka/ft/test.rs | 185 ++++++++- crates/feldera-types/Cargo.toml | 1 + crates/feldera-types/src/transport/kafka.rs | 383 +++++++++++++++++- crates/pipeline-manager/src/api/main.rs | 2 + .../docs/connectors/sources/kafka.md | 62 +++ openapi.json | 83 ++++ .../platform/test_kafka_header_filter.py | 154 +++++++ 9 files changed, 919 insertions(+), 8 deletions(-) create mode 100644 python/tests/platform/test_kafka_header_filter.py diff --git a/Cargo.lock b/Cargo.lock index 3e1ac4b7b20..b6db2308c10 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5530,6 +5530,7 @@ dependencies = [ "regex", "serde", "serde_json", + "serde_yaml", "tempfile", "thiserror 2.0.18", "time", diff --git a/crates/adapters/src/transport/kafka/ft/input.rs b/crates/adapters/src/transport/kafka/ft/input.rs index b54d9c1bae6..eb2cfc41fcf 100644 --- a/crates/adapters/src/transport/kafka/ft/input.rs +++ b/crates/adapters/src/transport/kafka/ft/input.rs @@ -23,7 +23,9 @@ use feldera_adapterlib::transport::{ use feldera_sqllib::{ByteArray, SqlString, Timestamp, Variant}; use feldera_types::config::FtModel; use feldera_types::program_schema::Relation; -use feldera_types::transport::kafka::{KafkaInputConfig, KafkaStartFromConfig}; +use feldera_types::transport::kafka::{ + CompiledHeaderFilter, KafkaInputConfig, KafkaStartFromConfig, +}; use itertools::Itertools; use rdkafka::client::OAuthToken; use rdkafka::config::RDKafkaLogLevel; @@ -66,6 +68,10 @@ const METADATA_TIMEOUT: Duration = Duration::from_secs(10); // to the worker thread. const ERROR_BUFFER_SIZE: usize = 1000; +/// A Kafka message's headers borrowed as `(key, value)` pairs for header +/// filtering. A `None` value denotes a header present with a null value. +type HeaderPairs<'a> = SmallVec<[(&'a str, Option<&'a [u8]>); 8]>; + pub struct KafkaFtInputEndpoint { config: Arc, } @@ -380,6 +386,16 @@ impl KafkaFtInputReaderInner { }) .collect::>(); + // Compile the header filter once and share it across partition + // receivers. A malformed filter is already rejected by + // `KafkaInputConfig::validate`, so this normally cannot fail. + let header_filter = config + .header_filter + .as_ref() + .map(|filter| filter.compile()) + .transpose()? + .map(Arc::new); + // Split every partition away as its own separate queue. let mut receivers = BTreeMap::new(); @@ -405,6 +421,7 @@ impl KafkaFtInputReaderInner { queue, next_offset, &config, + header_filter.clone(), unparker, )); receivers.insert(partition, receiver.clone()); @@ -999,6 +1016,10 @@ struct PartitionReceiver { config: KafkaInputConfig, metadata_requested: bool, + /// Compiled header filter, shared across partitions. Messages that do not + /// match are dropped before parsing. `None` admits all messages. + header_filter: Option>, + /// The maximum message offset that we want to receive, used as follows: /// /// - `i64::MIN`, the initial value, disables receiving messages entirely. @@ -1041,6 +1062,7 @@ impl PartitionReceiver { queue: PartitionQueue, next_offset: i64, config: &KafkaInputConfig, + header_filter: Option>, unparker: Unparker, ) -> Self { let metadata_requested = config.metadata_requested(); @@ -1057,6 +1079,7 @@ impl PartitionReceiver { fatal_error: AtomicBool::new(false), config: config.clone(), metadata_requested, + header_filter, unparker, } } @@ -1112,6 +1135,27 @@ impl PartitionReceiver { self.next_offset.load(Ordering::Relaxed) } + /// Returns `true` if `message` satisfies the configured header filter, or if + /// no filter is configured. Messages for which this returns `false` are + /// dropped before parsing. + fn header_filter_admits(&self, message: &BorrowedMessage<'_>) -> bool { + let Some(filter) = &self.header_filter else { + return true; + }; + + // Borrow the message's headers as (key, value) pairs; nothing is copied. + let headers: HeaderPairs = match message.headers() { + Some(headers) => (0..headers.count()) + .map(|i| { + let header = headers.get(i); + (header.key, header.value) + }) + .collect(), + None => SmallVec::new(), + }; + filter.matches(&headers) + } + /// Create record metadata from Kafka message containing only properties specified in the connector config. fn create_metadata(&self, message: &BorrowedMessage<'_>) -> Option { if !self.metadata_requested { @@ -1189,6 +1233,16 @@ impl PartitionReceiver { let next_offset = self.next_offset(); if offset >= next_offset { self.next_offset.store(offset + 1, Ordering::Relaxed); + + // Drop messages rejected by the header filter before parsing, + // so they never enter the offset ranges or the step hash. + // This runs after advancing `next_offset` (the consume + // position still moves forward) and is re-applied identically + // on replay, keeping ranges and hashes deterministic. + if !self.header_filter_admits(&message) { + return; + } + let timestamp = message.timestamp().to_millis().unwrap_or(i64::MIN); let payload = message.payload().unwrap_or(&[]); let metadata = self.create_metadata(&message); diff --git a/crates/adapters/src/transport/kafka/ft/test.rs b/crates/adapters/src/transport/kafka/ft/test.rs index cb2425a83bd..c4cffda5424 100644 --- a/crates/adapters/src/transport/kafka/ft/test.rs +++ b/crates/adapters/src/transport/kafka/ft/test.rs @@ -136,14 +136,37 @@ fn create_reader( DummyInputReceiver, Box, ) { - let config = serde_json::from_value(json!({ - "name": "kafka_input", - "config": { + create_reader_config(topic, resume_info, synchronize_partitions, None) +} + +/// Like [`create_reader`], but with an optional `header_filter` added to the +/// connector configuration. +fn create_reader_config( + topic: &str, + resume_info: Option, + synchronize_partitions: bool, + header_filter: Option, +) -> ( + Box, + DummyInputReceiver, + Box, +) { + let mut inner_config = json!({ "topic": topic, "log_level": "debug", - "start_from": "earliest", - "synchronize_partitions": synchronize_partitions, - } + "start_from": "earliest", + "synchronize_partitions": synchronize_partitions, + }); + if let Some(header_filter) = header_filter { + inner_config + .as_object_mut() + .unwrap() + .insert("header_filter".to_string(), header_filter); + } + + let config = serde_json::from_value(json!({ + "name": "kafka_input", + "config": inner_config, })) .unwrap(); @@ -165,6 +188,23 @@ fn create_reader( (endpoint, receiver, reader) } +/// Drain and return the payloads flushed to `receiver` so far, in order. +fn take_flushed(receiver: &DummyInputReceiver) -> Vec { + mem::take(&mut *receiver.inner.flushed.lock().unwrap()) +} + +/// Build the `headers` argument for [`TestProducer::send_message`] from +/// `(key, value)` pairs, where a `None` value denotes a header with a null +/// value. +fn headers(pairs: &[(&str, Option<&[u8]>)]) -> Option>>> { + Some( + pairs + .iter() + .map(|(key, value)| (key.to_string(), value.map(|v| v.to_vec()))) + .collect(), + ) +} + #[test] fn single_input() { test_input("single_input_ft", &[10]); @@ -469,6 +509,139 @@ fn test_input(topic: &str, batch_sizes: &[u32]) { } } +/// A configured header filter drops non-matching messages before parsing, and +/// the drop decision is deterministic across replay: a fresh reader replaying +/// the checkpointed offset range reproduces exactly the admitted records. +/// +/// To confirm this test catches a regression, make `header_filter_admits` +/// always return `true`; the connector then buffers all 6 messages instead of +/// 3 and both the live and the replay expectations fail. +#[test] +fn test_input_header_filter() { + init_test_logger(); + + let topic = "kafka_header_filter_ft"; + let _kafka_resources = KafkaResources::create_topics(&[(topic, 1)]); + + // Admit only messages whose `keep` header equals `yes`. + let filter = json!({"header": {"name": "keep", "pattern": "yes"}}); + + let (endpoint, receiver, reader) = + create_reader_config(topic, None, false, Some(filter.clone())); + reader.extend(); + + // Offsets 0..6. `keep=yes` at 0, 2, 4 are admitted; 1, 3 (`keep=no`) and 5 + // (no header) are dropped. The trailing drop at offset 5 lies beyond the + // recorded range and must never be replayed. + let producer = TestProducer::new(); + producer.send_message(b"m0", topic, headers(&[("keep", Some(b"yes"))])); + producer.send_message(b"m1", topic, headers(&[("keep", Some(b"no"))])); + producer.send_message(b"m2", topic, headers(&[("keep", Some(b"yes"))])); + producer.send_message(b"m3", topic, headers(&[("keep", Some(b"no"))])); + producer.send_message(b"m4", topic, headers(&[("keep", Some(b"yes"))])); + producer.send_message(b"m5", topic, None); + + // Only the 3 admitted messages are buffered. + receiver.expect_buffering(3); + reader.queue(false); + + // The offset range spans the first through the last admitted offset (0..5); + // dropped offsets inside and beyond the range do not change it. + let metadata = Metadata { + offsets: vec![0..5], + }; + receiver.expect(vec![ConsumerCall::Extended { + num_records: 3, + metadata: serde_json::to_value(&metadata).unwrap(), + }]); + assert_eq!(take_flushed(&receiver), vec!["m0", "m2", "m4"]); + + drop(endpoint); + drop(receiver); + drop(reader); + + // Replay the recorded range with the same filter. The filter is re-applied + // during replay, so exactly the admitted records reappear, in order. + let (_endpoint, receiver, reader) = create_reader_config(topic, None, false, Some(filter)); + receiver.inner.drop_buffered.store(true, Ordering::Release); + reader.replay(serde_json::to_value(&metadata).unwrap(), RmpValue::Nil); + receiver.expect(vec![ConsumerCall::Replayed { num_records: 3 }]); + assert_eq!(take_flushed(&receiver), vec!["m0", "m2", "m4"]); +} + +/// A boolean header filter (`and`/`or`/`not`) drops the right messages in the +/// real connector, exercising leading, middle, and trailing drops, and its +/// decisions are re-applied identically on replay. The filter reads headers +/// directly, so it works even though `include_headers` is not set. +#[test] +fn test_input_header_filter_boolean() { + init_test_logger(); + + let topic = "kafka_header_filter_boolean_ft"; + let _kafka_resources = KafkaResources::create_topics(&[(topic, 1)]); + + // Admit iff (env is prod or staging) and not (skip == drop). + let filter = json!({ + "and": [ + {"or": [ + {"header": {"name": "env", "pattern": "prod"}}, + {"header": {"name": "env", "pattern": "staging"}} + ]}, + {"not": {"header": {"name": "skip", "pattern": "drop"}}} + ] + }); + + let (endpoint, receiver, reader) = + create_reader_config(topic, None, false, Some(filter.clone())); + reader.extend(); + + let producer = TestProducer::new(); + // 0: env=dev -> drop (fails `or`); leading drop + // 1: env=prod -> admit + // 2: env=staging, skip=drop -> drop (fails `not`); middle drop + // 3: env=staging -> admit + // 4: env=prod, skip=drop -> drop (fails `not`); trailing drop + producer.send_message(b"m0", topic, headers(&[("env", Some(b"dev"))])); + producer.send_message(b"m1", topic, headers(&[("env", Some(b"prod"))])); + producer.send_message( + b"m2", + topic, + headers(&[("env", Some(b"staging")), ("skip", Some(b"drop"))]), + ); + producer.send_message(b"m3", topic, headers(&[("env", Some(b"staging"))])); + producer.send_message( + b"m4", + topic, + headers(&[("env", Some(b"prod")), ("skip", Some(b"drop"))]), + ); + + // Admitted: offsets 1 and 3. + receiver.expect_buffering(2); + reader.queue(false); + + // The range starts at the first admitted offset (1) and ends after the last + // (3), skipping the leading drop at offset 0. + let metadata = Metadata { + offsets: vec![1..4], + }; + receiver.expect(vec![ConsumerCall::Extended { + num_records: 2, + metadata: serde_json::to_value(&metadata).unwrap(), + }]); + assert_eq!(take_flushed(&receiver), vec!["m1", "m3"]); + + drop(endpoint); + drop(receiver); + drop(reader); + + // Replaying the recorded range reproduces the admitted records. + let (_endpoint, receiver, reader) = create_reader_config(topic, None, false, Some(filter)); + receiver.inner.drop_buffered.store(true, Ordering::Release); + reader.replay(serde_json::to_value(&metadata).unwrap(), RmpValue::Nil); + receiver.expect(vec![ConsumerCall::Replayed { num_records: 2 }]); + assert_eq!(take_flushed(&receiver), vec!["m1", "m3"]); +} + #[derive(Debug, PartialEq)] enum ConsumerCall { ParseErrors, diff --git a/crates/feldera-types/Cargo.toml b/crates/feldera-types/Cargo.toml index b5c291dea00..61fe3a86446 100644 --- a/crates/feldera-types/Cargo.toml +++ b/crates/feldera-types/Cargo.toml @@ -40,3 +40,4 @@ clap = { workspace = true } [dev-dependencies] csv = { workspace = true } tempfile = { workspace = true } +serde_yaml = { workspace = true } diff --git a/crates/feldera-types/src/transport/kafka.rs b/crates/feldera-types/src/transport/kafka.rs index 281f06a4b68..e2aea8f5e47 100644 --- a/crates/feldera-types/src/transport/kafka.rs +++ b/crates/feldera-types/src/transport/kafka.rs @@ -1,4 +1,5 @@ -use anyhow::{Error as AnyError, Result as AnyResult}; +use anyhow::{Error as AnyError, Result as AnyResult, anyhow, bail}; +use regex::bytes::Regex; use serde::de::{Error, SeqAccess, Visitor}; use serde::{Deserialize, Deserializer, Serialize}; use serde_json::Value as JsonValue; @@ -162,6 +163,19 @@ pub struct KafkaInputConfig { /// the partition whose events have been completely processed. #[serde(default)] pub synchronize_partitions: bool, + + /// Drop incoming messages whose Kafka headers do not satisfy this filter. + /// + /// The filter is a boolean expression (`and`, `or`, `not`) over regular + /// expression tests on individual header values (see [`HeaderFilter`]). + /// Messages that do not satisfy the filter are dropped before parsing and + /// never reach the pipeline. When absent (the default), all messages are + /// admitted. + /// + /// Filtering is independent of `include_headers`: it works whether or not + /// the matched headers are also surfaced to SQL via `CONNECTOR_METADATA()`. + #[serde(default, with = "crate::serde_via_value")] + pub header_filter: Option, } impl KafkaInputConfig { @@ -185,6 +199,7 @@ impl KafkaInputConfig { include_offset: None, include_topic: None, synchronize_partitions: false, + header_filter: None, } } @@ -214,6 +229,162 @@ impl<'de> Deserialize<'de> for KafkaInputConfig { } } +/// Maximum nesting depth of a [`HeaderFilter`], to bound recursion during +/// compilation and evaluation. +const MAX_HEADER_FILTER_DEPTH: usize = 64; + +/// A boolean filter over the headers of a Kafka message. +/// +/// The Kafka input connector uses this to drop messages whose headers do not +/// satisfy a predicate. It is a tree of boolean operators (`and`, `or`, `not`) +/// whose leaves are regular expression tests on individual header values. It +/// serializes as an externally tagged JSON object, for example: +/// +/// ```json +/// { +/// "and": [ +/// { "header": { "name": "event-type", "pattern": "created|updated" } }, +/// { "not": { "header": { "name": "source", "pattern": "test-.*" } } } +/// ] +/// } +/// ``` +/// +/// This admits a message only if it has an `event-type` header valued exactly +/// `created` or `updated` and does not have a `source` header whose value +/// starts with `test-`. +#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize, ToSchema)] +#[serde(rename_all = "snake_case")] +pub enum HeaderFilter { + /// Leaf test: matches when the message has a header named + /// [`HeaderMatch::name`] whose value matches [`HeaderMatch::pattern`]. + Header(HeaderMatch), + + /// Conjunction: matches when every nested filter matches. Must have at + /// least one operand. + And(Vec), + + /// Disjunction: matches when at least one nested filter matches. Must have + /// at least one operand. + Or(Vec), + + /// Negation: matches when the nested filter does not match. + Not(Box), +} + +/// A leaf of a [`HeaderFilter`]: a regular expression tested against the value +/// of a named Kafka header. +#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize, ToSchema)] +pub struct HeaderMatch { + /// Name of the header to test, matched exactly against the header key. + pub name: String, + + /// Regular expression ([Rust `regex` crate + /// syntax](https://docs.rs/regex/latest/regex/#syntax)) tested against the + /// header value. + /// + /// The value is matched as raw bytes, so non-UTF-8 values and byte patterns + /// work. The pattern must match the *entire* value: it is anchored + /// automatically, so `^`/`$` are unnecessary (though harmless). A header + /// present with a null value is matched as an empty byte sequence; a header + /// that appears more than once matches if any of its values match. + pub pattern: String, +} + +impl HeaderFilter { + /// Validate the filter without retaining the compiled form: rejects empty + /// `and`/`or`, nesting deeper than `MAX_HEADER_FILTER_DEPTH`, and invalid + /// leaf patterns. + pub fn validate(&self) -> AnyResult<()> { + self.compile().map(|_| ()) + } + + /// Compile into a [`CompiledHeaderFilter`] ready for evaluation, compiling + /// each leaf regular expression exactly once. Returns an error for the same + /// conditions as [`HeaderFilter::validate`]. + pub fn compile(&self) -> AnyResult { + Ok(CompiledHeaderFilter(self.compile_node(1)?)) + } + + fn compile_node(&self, depth: usize) -> AnyResult { + if depth > MAX_HEADER_FILTER_DEPTH { + bail!("Kafka header filter is nested more than {MAX_HEADER_FILTER_DEPTH} levels deep"); + } + match self { + HeaderFilter::Header(HeaderMatch { name, pattern }) => { + // Anchor to the whole value so the pattern matches the entire + // header value rather than a substring. The non-capturing group + // keeps any top-level alternation inside `pattern` bounded. + let regex = Regex::new(&format!(r"\A(?:{pattern})\z")).map_err(|error| { + anyhow!( + "invalid regular expression {pattern:?} for Kafka header {name:?}: {error}" + ) + })?; + Ok(CompiledNode::Header { + name: name.clone(), + regex, + }) + } + HeaderFilter::And(children) => { + if children.is_empty() { + bail!("Kafka header filter 'and' must have at least one operand"); + } + Ok(CompiledNode::And(Self::compile_children(children, depth)?)) + } + HeaderFilter::Or(children) => { + if children.is_empty() { + bail!("Kafka header filter 'or' must have at least one operand"); + } + Ok(CompiledNode::Or(Self::compile_children(children, depth)?)) + } + HeaderFilter::Not(child) => { + Ok(CompiledNode::Not(Box::new(child.compile_node(depth + 1)?))) + } + } + } + + fn compile_children(children: &[HeaderFilter], depth: usize) -> AnyResult> { + children.iter().map(|c| c.compile_node(depth + 1)).collect() + } +} + +/// A [`HeaderFilter`] compiled for evaluation. Holds the leaf regular +/// expressions compiled once so that matching a message allocates nothing. +#[derive(Debug)] +pub struct CompiledHeaderFilter(CompiledNode); + +#[derive(Debug)] +enum CompiledNode { + Header { name: String, regex: Regex }, + And(Vec), + Or(Vec), + Not(Box), +} + +impl CompiledHeaderFilter { + /// Evaluate the filter against a message's headers. + /// + /// `headers` lists the message's `(key, value)` pairs in order. A `None` + /// value denotes a header present with a null value and is matched as an + /// empty byte sequence. Returns `true` if the message satisfies the filter + /// and should be admitted. + pub fn matches(&self, headers: &[(&str, Option<&[u8]>)]) -> bool { + self.0.matches(headers) + } +} + +impl CompiledNode { + fn matches(&self, headers: &[(&str, Option<&[u8]>)]) -> bool { + match self { + CompiledNode::Header { name, regex } => headers + .iter() + .any(|&(key, value)| key == name.as_str() && regex.is_match(value.unwrap_or(b""))), + CompiledNode::And(children) => children.iter().all(|c| c.matches(headers)), + CompiledNode::Or(children) => children.iter().any(|c| c.matches(headers)), + CompiledNode::Not(child) => !child.matches(headers), + } + } +} + /// Where to begin reading a Kafka topic. #[derive(Debug, Clone, Default, Eq, PartialEq, Deserialize, Serialize, ToSchema)] #[serde(rename_all = "snake_case")] @@ -340,6 +511,13 @@ impl KafkaInputConfig { ) } + // Reject malformed filters (empty `and`/`or`, excessive nesting, or an + // invalid regular expression) at configuration time rather than when the + // connector starts. + if let Some(header_filter) = &self.header_filter { + header_filter.validate()?; + } + Ok(()) } } @@ -608,6 +786,8 @@ mod compat { include_topic: Option, #[serde(default)] synchronize_partitions: bool, + #[serde(default, with = "crate::serde_via_value")] + header_filter: Option, } impl TryFrom for super::KafkaInputConfig { @@ -687,7 +867,208 @@ mod compat { include_offset: compat.include_offset, include_topic: compat.include_topic, synchronize_partitions: compat.synchronize_partitions, + header_filter: compat.header_filter, }) } } } + +#[cfg(test)] +mod header_filter_tests { + use super::{HeaderFilter, KafkaInputConfig, MAX_HEADER_FILTER_DEPTH}; + use std::collections::BTreeMap; + + /// Compile the filter described by `json` and evaluate it against `headers`. + fn admits(json: &str, headers: &[(&str, Option<&[u8]>)]) -> bool { + let filter: HeaderFilter = serde_json::from_str(json).unwrap(); + filter.compile().unwrap().matches(headers) + } + + #[test] + fn leaf_matches_whole_value() { + let f = r#"{"header": {"name": "type", "pattern": "created|updated"}}"#; + assert!(admits(f, &[("type", Some(&b"created"[..]))])); + assert!(admits(f, &[("type", Some(&b"updated"[..]))])); + + // Patterns are anchored to the whole value: a substring must not match. + assert!(!admits(f, &[("type", Some(&b"created-x"[..]))])); + assert!(!admits(f, &[("type", Some(&b"x-created"[..]))])); + + // Wrong value, wrong header name, and missing header all fail a leaf. + assert!(!admits(f, &[("type", Some(&b"deleted"[..]))])); + assert!(!admits(f, &[("other", Some(&b"created"[..]))])); + assert!(!admits(f, &[])); + } + + #[test] + fn null_value_is_empty_bytes() { + let empty = r#"{"header": {"name": "h", "pattern": ""}}"#; + let nonempty = r#"{"header": {"name": "h", "pattern": ".+"}}"#; + + // A header with a null value matches an empty pattern but not `.+`. + assert!(admits(empty, &[("h", None)])); + assert!(!admits(nonempty, &[("h", None)])); + assert!(admits(empty, &[("h", Some(&b""[..]))])); + } + + #[test] + fn duplicate_header_any_value_matches() { + let f = r#"{"header": {"name": "k", "pattern": "b"}}"#; + assert!(admits(f, &[("k", Some(&b"a"[..])), ("k", Some(&b"b"[..]))])); + assert!(!admits( + f, + &[("k", Some(&b"a"[..])), ("k", Some(&b"c"[..]))] + )); + } + + #[test] + fn non_utf8_value_bytes() { + // A single-byte match against a non-UTF-8 value must work without panic. + let f = r#"{"header": {"name": "b", "pattern": "(?s-u:.)"}}"#; + assert!(admits(f, &[("b", Some(&[0xffu8][..]))])); + // Anchoring still applies: two bytes do not match a single-byte pattern. + assert!(!admits(f, &[("b", Some(&[0xffu8, 0xfe][..]))])); + } + + #[test] + fn boolean_combinations() { + // A AND (B OR C). + let f = r#"{"and": [ + {"header": {"name": "a", "pattern": "1"}}, + {"or": [ + {"header": {"name": "b", "pattern": "1"}}, + {"header": {"name": "c", "pattern": "1"}} + ]} + ]}"#; + assert!(admits(f, &[("a", Some(&b"1"[..])), ("b", Some(&b"1"[..]))])); + assert!(admits(f, &[("a", Some(&b"1"[..])), ("c", Some(&b"1"[..]))])); + assert!(!admits(f, &[("a", Some(&b"1"[..]))])); // neither b nor c + assert!(!admits(f, &[("b", Some(&b"1"[..]))])); // missing a + } + + #[test] + fn not_over_leaf() { + let f = r#"{"not": {"header": {"name": "env", "pattern": "prod"}}}"#; + assert!(!admits(f, &[("env", Some(&b"prod"[..]))])); // matching -> dropped + assert!(admits(f, &[("env", Some(&b"dev"[..]))])); // non-matching -> admitted + assert!(admits(f, &[])); // absent header -> admitted + } + + #[test] + fn de_morgan_equivalence() { + let lhs = r#"{"not": {"and": [ + {"header": {"name": "a", "pattern": "1"}}, + {"header": {"name": "b", "pattern": "1"}} + ]}}"#; + let rhs = r#"{"or": [ + {"not": {"header": {"name": "a", "pattern": "1"}}}, + {"not": {"header": {"name": "b", "pattern": "1"}}} + ]}"#; + let l = serde_json::from_str::(lhs) + .unwrap() + .compile() + .unwrap(); + let r = serde_json::from_str::(rhs) + .unwrap() + .compile() + .unwrap(); + for a in [None, Some(&b"1"[..])] { + for b in [None, Some(&b"1"[..])] { + let mut headers: Vec<(&str, Option<&[u8]>)> = Vec::new(); + if let Some(v) = a { + headers.push(("a", Some(v))); + } + if let Some(v) = b { + headers.push(("b", Some(v))); + } + assert_eq!( + l.matches(&headers), + r.matches(&headers), + "headers={headers:?}" + ); + } + } + } + + #[test] + fn rejects_empty_combinators() { + for json in [r#"{"and": []}"#, r#"{"or": []}"#] { + let filter: HeaderFilter = serde_json::from_str(json).unwrap(); + assert!(filter.validate().is_err(), "expected {json} to be rejected"); + } + } + + #[test] + fn rejects_deep_nesting() { + // Wrap a leaf in `not` more times than the depth limit allows. + let mut json = String::from(r#"{"header": {"name": "x", "pattern": "y"}}"#); + for _ in 0..MAX_HEADER_FILTER_DEPTH + 5 { + json = format!(r#"{{"not": {json}}}"#); + } + let filter: HeaderFilter = serde_json::from_str(&json).unwrap(); + assert!(filter.validate().is_err()); + } + + #[test] + fn rejects_invalid_regex() { + let filter: HeaderFilter = + serde_json::from_str(r#"{"header": {"name": "x", "pattern": "("}}"#).unwrap(); + assert!(filter.validate().is_err()); + } + + #[test] + fn serde_round_trip_external_tagging() { + let json = r#"{"and":[{"header":{"name":"t","pattern":"a|b"}},{"not":{"header":{"name":"s","pattern":"x"}}}]}"#; + let filter: HeaderFilter = serde_json::from_str(json).unwrap(); + let reserialized = serde_json::to_string(&filter).unwrap(); + assert_eq!(reserialized, json); + } + + #[test] + fn config_validate_rejects_bad_filter() { + let mut config = KafkaInputConfig::default(BTreeMap::new(), "topic"); + config.header_filter = Some(serde_json::from_str(r#"{"or": []}"#).unwrap()); + assert!(config.validate().is_err()); + } + + #[test] + fn config_deserializes_header_filter() { + // Exercises the compat deserialization path for the new field. + let config: KafkaInputConfig = serde_json::from_value(serde_json::json!({ + "topic": "t", + "bootstrap.servers": "localhost:9092", + "header_filter": {"header": {"name": "k", "pattern": "v"}} + })) + .unwrap(); + assert!(config.header_filter.is_some()); + } + + #[test] + fn config_with_header_filter_serializes_to_yaml() { + // Regression: the pipeline config is serialized to YAML when a pipeline + // is provisioned. `HeaderFilter` is a nested enum, which `serde_yaml` + // cannot serialize directly; `serde_via_value` keeps this working. + // Without it, the pipeline runner panics and the pipeline is stuck + // provisioning. + let mut config = KafkaInputConfig::default( + BTreeMap::from([( + "bootstrap.servers".to_string(), + "localhost:9092".to_string(), + )]), + "topic", + ); + config.header_filter = Some( + serde_json::from_str( + r#"{"and":[{"header":{"name":"a","pattern":"b"}},{"not":{"header":{"name":"c","pattern":"d"}}}]}"#, + ) + .unwrap(), + ); + + // Must not error (this is what panicked in the pipeline runner). + let yaml = serde_yaml::to_string(&config).expect("serialize config to YAML"); + + // And the config round-trips back through YAML. + let reparsed: KafkaInputConfig = serde_yaml::from_str(&yaml).unwrap(); + assert_eq!(reparsed.header_filter, config.header_filter); + } +} diff --git a/crates/pipeline-manager/src/api/main.rs b/crates/pipeline-manager/src/api/main.rs index 328ba13c7b3..462a731792a 100644 --- a/crates/pipeline-manager/src/api/main.rs +++ b/crates/pipeline-manager/src/api/main.rs @@ -385,6 +385,8 @@ It contains the following fields: feldera_types::transport::kafka::KafkaOutputConfig, feldera_types::transport::kafka::KafkaOutputFtConfig, feldera_types::transport::kafka::KafkaStartFromConfig, + feldera_types::transport::kafka::HeaderFilter, + feldera_types::transport::kafka::HeaderMatch, feldera_types::transport::nats::Auth, feldera_types::transport::nats::ConnectOptions, feldera_types::transport::nats::ConsumerConfig, diff --git a/docs.feldera.com/docs/connectors/sources/kafka.md b/docs.feldera.com/docs/connectors/sources/kafka.md index f8e98b55998..e9657d18448 100644 --- a/docs.feldera.com/docs/connectors/sources/kafka.md +++ b/docs.feldera.com/docs/connectors/sources/kafka.md @@ -24,6 +24,7 @@ tolerance](/pipelines/fault-tolerance). | `include_offset` | boolean | false | Whether to include Kafka offset name in connector metadata (see [Accessing Kafka metadata](#metadata)). | | `include_timestamp` | boolean | false | Whether to include Kafka timestamp in connector metadata (see [Accessing Kafka metadata](#metadata)). | | `synchronize_partitions` | boolean | false | Whether to read records in order of Kafka timestamp across partitions (see [Synchronizing partitions](#synchronizing-partitions)) | +| `header_filter` | filter | | Drop messages whose Kafka headers do not match a boolean expression of regular expressions (see [Filtering messages by header](#header-filter)). | The connector passes additional options directly to [**librdkafka**](https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md). Some of the relevant options: @@ -353,6 +354,67 @@ select from t; ``` +## Filtering messages by header + +The `header_filter` option drops incoming messages whose Kafka headers do not +match a filter. Dropped messages are discarded before parsing and never reach +the pipeline. When `header_filter` is omitted, all messages are admitted. + +A filter is a boolean expression whose leaves test individual header values +against regular expressions. Each node is a JSON object with a single key: + +| Node | Value | Matches when | +|------------|----------------------|-------------------------------------------------| +| `header` | `{name, pattern}` | a header named `name` has a value matching `pattern` | +| `and` | list of filters | every nested filter matches | +| `or` | list of filters | at least one nested filter matches | +| `not` | a filter | the nested filter does not match | + +The `header` leaf takes two fields: + +* `name`: the header name, matched exactly against the header key. +* `pattern`: a [regular expression](https://docs.rs/regex/latest/regex/#syntax) + tested against the header value. The value is matched as raw bytes, so + non-UTF-8 values work. The pattern must match the **entire** value (it is + anchored automatically), so `^` and `$` are unnecessary. + +Leaf matching details: + +* A message with no header of the given `name` does not match the leaf. +* A header present with a null value is matched as an empty byte sequence. +* If a header appears more than once, the leaf matches when any of its values + match. + +Filtering is independent of [`include_headers`](#metadata): a header can be used +in a filter whether or not it is also exposed to SQL through +`CONNECTOR_METADATA()`. + +### Examples + +Admit only messages whose `event-type` header is exactly `created` or +`updated`: + +```json +"header_filter": { + "header": { "name": "event-type", "pattern": "created|updated" } +} +``` + +Admit messages from `prod` or `staging`, unless they carry a `skip` header set +to `true`: + +```json +"header_filter": { + "and": [ + { "or": [ + { "header": { "name": "env", "pattern": "prod" } }, + { "header": { "name": "env", "pattern": "staging" } } + ]}, + { "not": { "header": { "name": "skip", "pattern": "true" } } } + ] +} +``` + ## Tolerating missing data on resume The `resume_earliest_if_data_expires` setting controls how the Kafka diff --git a/openapi.json b/openapi.json index bf05deaf064..b29f8a7a19e 100644 --- a/openapi.json +++ b/openapi.json @@ -9709,6 +9709,81 @@ } } }, + "HeaderFilter": { + "oneOf": [ + { + "type": "object", + "required": [ + "header" + ], + "properties": { + "header": { + "$ref": "#/components/schemas/HeaderMatch" + } + } + }, + { + "type": "object", + "required": [ + "and" + ], + "properties": { + "and": { + "type": "array", + "items": { + "$ref": "#/components/schemas/HeaderFilter" + }, + "description": "Conjunction: matches when every nested filter matches. Must have at\nleast one operand." + } + } + }, + { + "type": "object", + "required": [ + "or" + ], + "properties": { + "or": { + "type": "array", + "items": { + "$ref": "#/components/schemas/HeaderFilter" + }, + "description": "Disjunction: matches when at least one nested filter matches. Must have\nat least one operand." + } + } + }, + { + "type": "object", + "required": [ + "not" + ], + "properties": { + "not": { + "$ref": "#/components/schemas/HeaderFilter" + } + } + } + ], + "description": "A boolean filter over the headers of a Kafka message.\n\nThe Kafka input connector uses this to drop messages whose headers do not\nsatisfy a predicate. It is a tree of boolean operators (`and`, `or`, `not`)\nwhose leaves are regular expression tests on individual header values. It\nserializes as an externally tagged JSON object, for example:\n\n```json\n{\n\"and\": [\n{ \"header\": { \"name\": \"event-type\", \"pattern\": \"created|updated\" } },\n{ \"not\": { \"header\": { \"name\": \"source\", \"pattern\": \"test-.*\" } } }\n]\n}\n```\n\nThis admits a message only if it has an `event-type` header valued exactly\n`created` or `updated` and does not have a `source` header whose value\nstarts with `test-`." + }, + "HeaderMatch": { + "type": "object", + "description": "A leaf of a [`HeaderFilter`]: a regular expression tested against the value\nof a named Kafka header.", + "required": [ + "name", + "pattern" + ], + "properties": { + "name": { + "type": "string", + "description": "Name of the header to test, matched exactly against the header key." + }, + "pattern": { + "type": "string", + "description": "Regular expression ([Rust `regex` crate\nsyntax](https://docs.rs/regex/latest/regex/#syntax)) tested against the\nheader value.\n\nThe value is matched as raw bytes, so non-UTF-8 values and byte patterns\nwork. The pattern must match the *entire* value: it is anchored\nautomatically, so `^`/`$` are unnecessary (though harmless). A header\npresent with a null value is matched as an empty byte sequence; a header\nthat appears more than once matches if any of its values match." + } + } + }, "HealthStatus": { "type": "object", "required": [ @@ -10058,6 +10133,14 @@ "description": "Maximum timeout in seconds to wait for the endpoint to join the Kafka\nconsumer group during initialization.", "minimum": 0 }, + "header_filter": { + "allOf": [ + { + "$ref": "#/components/schemas/HeaderFilter" + } + ], + "nullable": true + }, "include_headers": { "type": "boolean", "description": "Whether to include Kafka headers in the record metadata.\n\nWhen `true`, Kafka message headers are available via the `CONNECTOR_METADATA()` function.\nSee for details.", diff --git a/python/tests/platform/test_kafka_header_filter.py b/python/tests/platform/test_kafka_header_filter.py new file mode 100644 index 00000000000..603481a69ff --- /dev/null +++ b/python/tests/platform/test_kafka_header_filter.py @@ -0,0 +1,154 @@ +"""End-to-end tests for the Kafka input connector's header filter. + +Each test creates a pipeline whose Kafka input connector carries a +``header_filter``, produces messages with various headers, and checks that only +the messages satisfying the filter reach the table. A final admitted +"sentinel" message (produced last on a single-partition topic, so it is +processed last) marks completeness: once it appears, every earlier message has +been decided, so the observed row set is final. +""" + +import json +import uuid +from typing import Any, Optional + +from confluent_kafka import Producer +from confluent_kafka.admin import AdminClient, NewTopic +from feldera import Pipeline, PipelineBuilder +from tests import KAFKA_BOOTSTRAP, TEST_CLIENT +from tests.platform.helper import wait_for_condition + +SENTINEL_ID = 100 + + +def _random_topic(prefix: str) -> str: + return f"{prefix}-{uuid.uuid4().hex[:12]}" + + +def _create_topic(admin: AdminClient, topic: str) -> None: + futures = admin.create_topics( + [NewTopic(topic=topic, num_partitions=1, replication_factor=1)] + ) + futures[topic].result(timeout=30) + + +def _delete_topic_best_effort(admin: AdminClient, topic: str) -> None: + try: + futures = admin.delete_topics([topic], operation_timeout=10) + futures[topic].result(timeout=10) + except Exception: + # Topic deletion can be disabled on some brokers; cleanup is best-effort. + pass + + +def _produce( + topic: str, records: list[tuple[int, Optional[list[tuple[str, bytes]]]]] +) -> None: + """Produce ``{"id": }`` messages, each with the given headers (or none).""" + producer = Producer({"bootstrap.servers": KAFKA_BOOTSTRAP}) + for record_id, headers in records: + producer.produce( + topic, + value=json.dumps({"id": record_id}).encode("utf-8"), + headers=headers, + ) + remaining = producer.flush(timeout=30) + assert remaining == 0, f"failed to flush Kafka messages, remaining={remaining}" + + +def _run_filter_test( + pipeline_name: str, + header_filter: dict[str, Any], + records: list[tuple[int, Optional[list[tuple[str, bytes]]]]], + expected_ids: set[int], +) -> None: + admin = AdminClient({"bootstrap.servers": KAFKA_BOOTSTRAP}) + topic = _random_topic("header-filter-in") + _create_topic(admin, topic) + + input_connector = { + "name": "kafka_in", + "transport": { + "name": "kafka_input", + "config": { + "topic": topic, + "bootstrap.servers": KAFKA_BOOTSTRAP, + "start_from": "earliest", + "header_filter": header_filter, + }, + }, + "format": {"name": "json", "config": {"update_format": "raw", "array": False}}, + } + + sql = f""" + CREATE TABLE input_t(id INT) WITH ( + 'connectors' = '{json.dumps([input_connector])}' + ); + CREATE MATERIALIZED VIEW output_v AS SELECT * FROM input_t; + """.strip() + + pipeline: Pipeline = PipelineBuilder( + TEST_CLIENT, name=pipeline_name, sql=sql + ).create_or_replace() + pipeline.start() + + def ingested_ids() -> set[int]: + return {row["id"] for row in pipeline.query("SELECT id FROM output_v")} + + try: + # The sentinel is produced last; the filter must admit it. + assert SENTINEL_ID in expected_ids + _produce(topic, records) + + wait_for_condition( + "sentinel message ingested", + lambda: SENTINEL_ID in ingested_ids(), + timeout_s=60.0, + poll_interval_s=1.0, + ) + + # The sentinel is last on a single-partition topic, so ingestion is now + # complete: the observed set is final and must equal the expected set. + assert ingested_ids() == expected_ids + finally: + pipeline.stop(force=True) + _delete_topic_best_effort(admin, topic) + + +def test_kafka_header_filter_leaf(pipeline_name): + """A single regex leaf admits only whole-value matches; anchoring and + missing headers behave as documented.""" + header_filter = {"header": {"name": "event", "pattern": "created|updated"}} + records = [ + (1, [("event", b"created")]), # admit + (2, [("event", b"deleted")]), # drop: wrong value + (3, [("event", b"updated")]), # admit + (4, None), # drop: header absent + (5, [("event", b"created-x")]), # drop: anchored, no substring match + (SENTINEL_ID, [("event", b"created")]), # admit (sentinel) + ] + _run_filter_test(pipeline_name, header_filter, records, {1, 3, SENTINEL_ID}) + + +def test_kafka_header_filter_boolean(pipeline_name): + """A boolean filter combines several header tests with and/or/not.""" + header_filter = { + "and": [ + { + "or": [ + {"header": {"name": "env", "pattern": "prod"}}, + {"header": {"name": "env", "pattern": "staging"}}, + ] + }, + {"not": {"header": {"name": "skip", "pattern": "true"}}}, + ] + } + records = [ + (1, [("env", b"prod")]), # admit + (2, [("env", b"dev")]), # drop: fails `or` + (3, [("env", b"staging"), ("skip", b"false")]), # admit: skip != "true" + (4, [("env", b"prod"), ("skip", b"true")]), # drop: fails `not` + (5, [("env", b"staging")]), # admit + (SENTINEL_ID, [("env", b"prod")]), # admit (sentinel) + ] + _run_filter_test(pipeline_name, header_filter, records, {1, 3, 5, SENTINEL_ID})