[adapters] Kafka header filters.#6607
Conversation
mythical-fred
left a comment
There was a problem hiding this comment.
LGTM. One tiny formatting nit inline; nothing blocking.
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 <ryzhyk@gmail.com>
ca95793 to
f54cbcc
Compare
|
This is the kind of stuff that looks like it could live well in a preprocessor. |
When you have a hammer... This is true, but we have users who'd like to use this with multiple data formats. They'd have to write multiple Rust preprocessors. A connector config setting is easier. |
| [dev-dependencies] | ||
| csv = { workspace = true } | ||
| tempfile = { workspace = true } | ||
| serde_yaml = { workspace = true } |
|
|
||
| // Admit iff (env is prod or staging) and not (skip == drop). | ||
| let filter = json!({ | ||
| "and": [ |
There was a problem hiding this comment.
We should standardize on the language for these expressions, in the push-down PR we are talking about JsonLogic https://crates.io/crates/jsonlogic-rs
But JsonLogic does not have regexes, these are very difficult across languages... sigh
| { "not": { "header": { "name": "skip", "pattern": "true" } } } | ||
| ] | ||
| } | ||
| ``` |
There was a problem hiding this comment.
Could use an example with escaped regex characters, so one does not have to read the regex documentation
| against regular expressions. Each node is a JSON object with a single key: | ||
|
|
||
| | Node | Value | Matches when | | ||
| |------------|----------------------|-------------------------------------------------| |
There was a problem hiding this comment.
Not very fond of this language, let's make this as general as possible, we may reuse it in several places.
There should be a crate for parsing and validating this language.
"header" is fine, and it should return a string.
Use "pattern" as a function which takes a string and returns a regex
Use "match" as a function which takes a string and a regex
There was a problem hiding this comment.
Not very fond of this language, let's make this as general as possible,
why?
we may reuse it in several places.
what do you have in mind?
"header" is fine, and it should return a string.
Use "pattern" as a function which takes a string and returns a regex
Use "match" as a function which takes a string and a regex
I don't understand what this means. Is this some JsonLogic stuff? Seems like we're trying to massively overengineer this.
There was a problem hiding this comment.
Doesn't seem like a good idea to force users to learn JsonLogic to write a simple header filter...
There was a problem hiding this comment.
I disagree, they will have to learn SOME language, let's at least make it a consistent language
|
|
||
| /// Drop incoming messages whose Kafka headers do not satisfy this filter. | ||
| /// | ||
| /// The filter is a boolean expression (`and`, `or`, `not`) over regular |
There was a problem hiding this comment.
Do not describe here the filter language at all.
| /// starts with `test-`. | ||
| #[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize, ToSchema)] | ||
| #[serde(rename_all = "snake_case")] | ||
| pub enum HeaderFilter { |
There was a problem hiding this comment.
Move this to a separate crate, something like JsonExpression.
|
Sorry to make your life harder, but I think a more general solution is worth pursuing. We already have a myriad little languages sprinkled everywhere. |
|
I can help implementing the "JsonExpression" crate I mentioned if it helps. Or maybe we can reuse some of the stuff that @gz has built already. |
|
there is a query language for json built by a friend of mine https://www.jsoniq.org but I agree lets keep it simple for now |
mihaibudiu
left a comment
There was a problem hiding this comment.
This design is fine if we expect that the filter language is complete in its current form.
New Kafka input connector config setting to filter messages based on Kafka message headers. Supports primitive checks against regular expressions and arbitrary Boolean combinations.
Describe Manual Test Plan
Tested with a single-table pipeline and a Kafka topic with 5 messages, and a couple of headers each.
Checklist
Breaking Changes?
Mark if you think the answer is yes for any of these components:
Describe Incompatible Changes