Kafka Connect Standalone：ConnectException：找不到任何实现连接器且名称与 ElasticsearchSinkConnector 匹配的类

Question

我正在尝试在 EC2 上运行 kafka connect 并将数据从 MSK 发送到 Elastic Search。 这就是我正在做的。

sudo yum install java-1.8.0
wget https://archive.apache.org/dist/kafka/2.2.1/kafka_2.12-2.2.1.tgz
tar -xzf kafka_2.12-2.2.1.tgz

Kafka连接安装

wget /usr/local http://packages.confluent.io/archive/5.2/confluent-5.2.0-2.11.tar.gz -P ~/Downloads/
tar -zxvf ~/Downloads/confluent-5.2.0-2.11.tar.gz -C ~/Downloads/
sudo mv ~/Downloads/confluent-5.2.0 /usr/local/confluent

我修改了两个属性文件

vim /usr/local/confluent/etc/kafka-connect-elasticsearch/quickstart-elasticsearch.properties

这是我创建的主题名称为我创建的 kafka 主题和弹性搜索的连接 url

第二个属性文件

vim /usr/local/confluent/etc/kafka/connect-standalone.properties

这里我只修改了引导服务器 URL，给出了所有 3 个引导 URL

最后我像这样运行我的连接器

/usr/local/confluent/bin/connect-standalone /usr/local/confluent/etc/kafka/connect-standalone.properties /usr/local/confluent/etc/kafka-connect-elasticsearch/quickstart-elasticsearch.properties

然后我得到以下错误

[2019-12-30 20:35:38,109] INFO Kafka Connect standalone worker initialization took 3890ms (org.apache.kafka.connect.cli.ConnectStandalone:96)
[2019-12-30 20:35:38,109] INFO Kafka Connect starting (org.apache.kafka.connect.runtime.Connect:50)
[2019-12-30 20:35:38,109] INFO Herder starting (org.apache.kafka.connect.runtime.standalone.StandaloneHerder:87)
[2019-12-30 20:35:38,109] INFO Worker starting (org.apache.kafka.connect.runtime.Worker:162)
[2019-12-30 20:35:38,109] INFO Starting FileOffsetBackingStore with file /tmp/connect.offsets (org.apache.kafka.connect.storage.FileOffsetBackingStore:58)
[2019-12-30 20:35:38,111] INFO Worker started (org.apache.kafka.connect.runtime.Worker:167)
[2019-12-30 20:35:38,111] INFO Herder started (org.apache.kafka.connect.runtime.standalone.StandaloneHerder:89)
[2019-12-30 20:35:38,111] INFO Kafka Connect started (org.apache.kafka.connect.runtime.Connect:55)
[2019-12-30 20:35:38,113] ERROR Failed to create job for /usr/local/confluent/etc/kafka-connect-elasticsearch/quickstart-elasticsearch.properties (org.apache.kafka.connect.cli.ConnectStandalone:108)
[2019-12-30 20:35:38,113] ERROR Stopping after connector error (org.apache.kafka.connect.cli.ConnectStandalone:119)
java.util.concurrent.ExecutionException: org.apache.kafka.connect.errors.ConnectException: Failed to find any class that implements Connector and which name matches io.confluent.connect.elasticsearch.ElasticsearchSinkConnector, available connectors are: PluginDesc{klass=class org.apache.kafka.connect.file.FileStreamSinkConnector, name='org.apache.kafka.connect.file.FileStreamSinkConnector', version='2.2.0-cp1', encodedVersion=2.2.0-cp1, type=sink, typeName='sink', location='classpath'}, PluginDesc{klass=class org.apache.kafka.connect.file.FileStreamSourceConnector, name='org.apache.kafka.connect.file.FileStreamSourceConnector', version='2.2.0-cp1', encodedVersion=2.2.0-cp1, type=source, typeName='source', location='classpath'}, PluginDesc{klass=class org.apache.kafka.connect.tools.MockConnector, name='org.apache.kafka.connect.tools.MockConnector', version='2.2.0-cp1', encodedVersion=2.2.0-cp1, type=connector, typeName='connector', location='classpath'}, PluginDesc{klass=class org.apache.kafka.connect.tools.MockSinkConnector, name='org.apache.kafka.connect.tools.MockSinkConnector', version='2.2.0-cp1', encodedVersion=2.2.0-cp1, type=sink, typeName='sink', location='classpath'}, PluginDesc{klass=class org.apache.kafka.connect.tools.MockSourceConnector, name='org.apache.kafka.connect.tools.MockSourceConnector', version='2.2.0-cp1', encodedVersion=2.2.0-cp1, type=source, typeName='source', location='classpath'}, PluginDesc{klass=class org.apache.kafka.connect.tools.SchemaSourceConnector, name='org.apache.kafka.connect.tools.SchemaSourceConnector', version='2.2.0-cp1', encodedVersion=2.2.0-cp1, type=source, typeName='source', location='classpath'}, PluginDesc{klass=class org.apache.kafka.connect.tools.VerifiableSinkConnector, name='org.apache.kafka.connect.tools.VerifiableSinkConnector', version='2.2.0-cp1', encodedVersion=2.2.0-cp1, type=source, typeName='source', location='classpath'}, PluginDesc{klass=class org.apache.kafka.connect.tools.VerifiableSourceConnector, name='org.apache.kafka.connect.tools.VerifiableSourceConnector', version='2.2.0-cp1', encodedVersion=2.2.0-cp1, type=source, typeName='source', location='classpath'}

我对此处的 plugin.path 感到困惑。 我该如何设置？ 它带有 kafka-connect 还是我必须设置它？

Answer 1

我会指出，Logstash 也适用于将 Kafka 连接到 Elasticsearch

我对这里的 plugin.path 感到困惑。我该如何设置？是否自带kafka-connect

查看 connect-standalone.properties 的底部并阅读那里的 cmets，然后取消注释插件路径属性

https://github.com/apache/kafka/blob/trunk/config/connect-standalone.properties#L32-L41

正如我在这里回答的那样，最好不要使用 tarball；通过分别下载 Confluent Platform 和 Kafka，您已经复制了 Zookeeper 和 Kafka。
请使用 YUM 安装 Confluent Platform，其中包括 Zookeeper、Apache Kafka 和您的 Elasticsearch 连接器

Failed to find any class that implements Connector and which name matches io.confluent.connect.elasticsearch.ElasticsearchSinkConnector