Kafka Mysql CDC 到弹性搜索

Question

我试图通过 source-connector 将我的 MySql db 更改外包到 kafka 主题，并且有效。现在我想将这些数据发送到弹性搜索实例。

为此，我关注了这个Kafka Connect Elasticsearch: Consuming and Indexing with Kafka Connect 和这个Kafka Connect and Elasticsearch。

对于 mysql 到 kafka 的 CDC，我可以看到我在 mysql 中所做的更改并读取它创建一个源连接器，但是当我创建另一个连接器 elasticsearch-sink 连接器时，source-connectr task.state强> 显示Failed！因此，尽管索引是在 es-config.properties 文件中创建的，但数据库更改不会进入 ES。

我已将 jar/s 放在 kafka-dir 中，其中 source-connector 的 lib 工作（以避免有关类路径问题的进一步问题）。

创建elaticsearch-sink-connector 时出现此错误（当然，我没有错误，所有库都在同一个目录中！）：

连接器的错误插件类加载器： 'io.confluent.connect.elasticsearch.ElasticsearchSinkConnector' 是 未找到。返回： org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader@5cc126dc (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:165)

我通过这个来运行我的连接器：

bin/connect-standalone.sh config/connect-standalone.properties config/elasticsearch-connect.properties

简而言之，我的连接器 task.state 一次只保留一个 RUNNING

编辑：
plugin.path 用于 connect-standablone.properties 文件：

plugin.path=/media/***/projects/playground/kafka/kafka_2.12-2.4.0, /media/***/projects/playground/kafka/kafka-connect-elasticsearch/target/kafka-connect-elasticsearch-3.2.0-SNAPSHOT-package/share/java

它们都包含 es-connector jar。最后一个稍后添加但仍然相同

我现在该怎么办？

Answer 1

当我将 schema.enable 更改为 false 时，事情就像魅力一样发挥作用

key.converter.schemas.enable=false
value.converter.schemas.enable=false

并在plugin.path 之后添加了额外的/，尽管没有/ 它适用于源连接器！

编辑：我忘了提到我已经用 5.4.0 版本替换了我的连接器版本，正如 cricket_007 提到的那样

Edit-2： 后来我进行了更多调查，发现额外的/ 问题以及下面提到的新关键属性帮助我摆脱了连接器的FAILED 状态（一次只有一个连接器正在运行）：

internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=false

在 connect-standalone.properties 文件中

谢谢