Kafka Connect Debezium过滤器转换报错求助：op字段不存在导致任务异常终止

我来帮你拆解这个问题的核心原因，以及对应的修复方案：

根本原因 #

你遇到的问题本质是Debezium会生成两种完全不同结构的事件记录，而你的Filter规则没有区分它们：

• 第一种是业务数据变更记录：就是你期望捕获的CRUD操作，这类记录的Value里确实有op字段（'c'=创建、'u'=更新、'd'=删除）；
• 第二种是SchemaChange元数据事件：当数据库执行DDL操作（比如ALTER TABLE、CREATE TABLE）时，Debezium会生成这类事件来同步表结构变更，它的Value结构和数据变更记录完全不一样，根本没有op字段，对应的Schema类型是io.debezium.connector.sqlserver.SchemaChangeValue。

你的Filter转换会对所有流经连接器的记录执行value.op == 'c'这个表达式，一旦SchemaChange事件出现，表达式尝试访问不存在的op字段，就会直接抛出op is not a valid field name错误，导致连接器任务终止。

至于为什么重启/重建后几天才会复发？因为SchemaChange事件不是连接器启动时就有的，只有当数据库发生DDL操作时才会生成，所以刚开始运行时没有这类事件，连接器正常工作，后续遇到DDL操作就会触发报错。

修复方案 #

你需要修改Filter的条件表达式，先判断当前记录是否是包含op字段的数据变更记录，再进行过滤。这里提供几种可靠的写法：

方案1：检查记录是否包含op字段（最通用） #

把transforms.filter.condition改成：

value.containsKey('op') && value.op == 'c'

这个表达式会先判断Value中是否存在op字段，只有存在时才会继续判断op的值，完美避开SchemaChange事件的问题。

方案2：通过Schema类型精准判断 #

如果想更精准地排除SchemaChange事件，可以直接匹配它的Schema名称：

value.schema().name() != 'io.debezium.connector.sqlserver.SchemaChangeValue' && value.op == 'c'

这种方式直接针对SchemaChange事件的标识进行过滤，逻辑更清晰。

方案3：用Groovy安全导航简化写法 #

Groovy的安全导航操作符?.可以在字段不存在时返回null，从而避免报错：

value?.op == 'c'

当记录是SchemaChange事件时，value?.op会返回null，表达式结果为false，这类记录会被自动过滤掉，同时不会触发错误，写法最简洁。

验证修改后的创建命令（以方案1为例） #

修改后的连接器创建命令需要注意单引号的转义（避免JSON解析失败）：

curl -X POST "${KAFKA_CONNECT_HOST}/connectors" -H "Content-Type: application/json" -d '{ "name": "DebeziumSMS", "config": { "connector.class": "io.debezium.connector.sqlserver.SqlServerConnector", "tasks.max": 1, "database.hostname": "bill-srv02.corp.oblakowifi.ru", "database.port": 1433, "database.user": "sa", "database.password": "******", "database.dbname": "sms", "database.server.name": "server-test", "database.history.kafka.bootstrap.servers": "192.168.26.142:9092", "database.history.kafka.topic": "schema-changes.sms", "errors.log.enable": "true", "database.history.skip.unparseable.ddl": "true", "time.precision.mode": "connect", "transforms": "filter", "transforms.filter.type": "io.debezium.transforms.Filter", "transforms.filter.language": "jsr223.groovy", "transforms.filter.condition": "value.containsKey('\''op'\'') && value.op == '\''c'\''" } }'

这样修改后，即使数据库发生DDL操作产生SchemaChange事件，Filter也会跳过这些记录，不会触发报错，同时正常保留创建操作的数据变更记录。

内容的提问来源于stack exchange，提问作者S.Pot