Kafka 连接器配置错误：filter.condition：定义的 json 路径无效答案

【问题标题】：Kafka connector config error: filter.condition: Invalid json path definedKafka 连接器配置错误：filter.condition：定义的 json 路径无效
【发布时间】：2020-11-04 06:30:06
【问题描述】：

我正在尝试使用 Confluent 的 Filter SMT 和 Debezium 示例 unwrap-smt。

我在源连接器 (Debezium MySQL) 配置中添加了以下配置：

    "transforms": "route,csFilter",
    ...
    ...
    "transforms.csFilter.type": "io.confluent.connect.transforms.Filter$Value",
    "transforms.csFilter.filter.condition": "$.payload.after.source == 2",
    "transforms.csFilter.filter.type": "exclude",
    "transforms.csFilter.missing.or.null.behavior": "fail"

由于这个 Filter SMT 是 Confluent 提供的，我下载了 jar file 并将 (connect-transforms, connect-utils, json-path) jar 文件复制到 path-to-kafka/connect/debezium-connector-mysql 目录。

当我尝试注册 Debezium MySQL 源连接器时，

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" 
localhost:8083/connectors/ -d @source_connector_config.json

我收到了这个错误：

{"error_code":400,
"message":"Connector configuration is invalid and contains the following 1 error(s):\n
Invalid value $.payload.after.source == 2 for configuration filter.condition: Invalid json path defined. 
Please refer to https://github.com/json-path/JsonPath README for correct use of json path.\n
You can also find the above list of errors at the endpoint `/{connectorType}/config/validate`"}

我使用this guide 中提供的示例检查了 JSON 路径表达式。好像没问题。

你能告诉我正确的方向吗？我错过了什么？谢谢。

【问题讨论】：

试试这个条件：$.payload.after[?(@.source == 2)]
成功了。 JSON 数据的“后”属性没有任何列表。你能解释一下为什么我们在 json 路径表达式中有下标运算符 [] 吗？另外，如果您可以将解决方案复制为答案，我将接受它作为答案。谢谢！
请查看docs：过滤表达式需要括号。示例：baeldung.com/…

标签： apache-kafka apache-kafka-connect confluent-platform jsonpath debezium

【解决方案1】：

请尝试使用此条件：$.payload.after[?(@.source == 2)]

【讨论】：