基于debezium内容的路由配置

Question

我使用的是confluent，所以我已经根据confluent文档安装了dibezium连接器，在connect.properties中使用confluent-hub我确实有入口

plugin.path=/usr/share/java,/opt/confluent-6.0.0/share/confluent-hub-components

我需要使用io.debezium.transforms.ContentBasedRouter https://debezium.io/documentation/reference/1.3/configuration/content-based-routing.html

因此，根据debezium文档，我已经下载了debezium-scripting-1.3.1.Final.jar并将其放入/opt/confluent-6.0.0/share/confluent-hub-components/，并将其复制到/opt/confluent-6.0.0/share/confluent-hub-components/debezium-debezium-connector-sqlserver/lib目录中，这里是我的mysql_src.json连接器中的条目

"transforms": "unwrap,route",
"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
"transforms.unwrap.add.fields": "source.snapshot",
"transforms.route.type": "io.debezium.transforms.ContentBasedRouter",
"transforms.route.language": "jsr223.groovy",
"transforms.route.topic.expression": "value.__source_snapshot == 'false' ? 'test'"

当我尝试配置/加载此连接器时，收到以下错误消息

[2020-12-15 22:18:45,351] ERROR [Worker clientId=connect-1, groupId=connect-cluster] Failed to reconfigure connector's tasks, retrying after backoff: (org.apache.kafka.connect.runtime.distributed.DistributedHerder:1369)
java.lang.NoClassDefFoundError: io/debezium/DebeziumException

有什么建议可以解决这个问题吗？

Answer 1

根据docs，您还需要获得一个JSR-223脚本引擎实现，并将其内容添加到Kafka Connect环境的Debezium插件目录中，因为：

JSR223API的任何实现都没有附带JSRDebezium.要在Debezium中使用表达式语言，必须下载该语言的JSR223脚本引擎实现，并将该语言实现使用的任何其他JAR文件添加到Debezium连接器插件目录中。

Answer 2

我不确定配置是正确的，但我通过了第一个配置问题(我希望)我现在面临着另一个问题，我将在不同的问题中描述。我不确定出了什么问题，我做了以下事情

在分布式模式下使用命令行启动/停止脚本(不使用confluent cli)清理zookeeper directories

• Clean up kafka directories

• Run kafka这解决了

的java.lang.NoClassDefFoundError: io/debezium/DebeziumException错误