我试图从运行的汇合插件kafka-connect-elasticsearch为了从卡夫卡(V0.11.0.1)直接进入Elasticsearch流主题(没有把Logstash之间)。卡夫卡连接Elasticsearch - 的NoSuchMethodError
我建立使用Maven连接器 -
$ cd kafka-connect-elasticsearch
$ mvn clean package
然后我创建了需要的配置文件 -
name=es-cluster-lab
connector.class=io.confluent.connect.elasticsearch.ElasticsearchSinkConnector
tasks.max=1
topics=filebeats-test
topic.index.map=filebeats-test:kafka_test_index
key.ignore=true
schema-ignore=true
connection.url=http://elastic:9200
type.name=log
根据新卡夫卡的Classpath隔离规范,我还添加了下面一行到我connect-standalone.properties
文件 -
plugin.path=/home/kafka/kafka-connect-elasticsearch-3.3.0/target/kafka-connect-elasticsearch-3.3.0-development/share/java/kafka-connect-elasticsearch/
我去运行脚本...
bin/connect-standalone.sh config/connect-standalone.properties config/elasticsearch-connect.properties
...并收到下面的错误。
[2017-09-14 16:08:26,599] INFO Loading plugin from: /home/kafka/kafka-connect-elasticsearch-3.3.0/target/kafka-connect-elasticsearch-3.3.0-development/share/java/kafka-connect-elasticsearch/slf4j-api-1.7.25.jar (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:176)
Exception in thread "main" java.lang.NoSuchMethodError: com.google.common.collect.Sets$SetView.iterator()Lcom/google/common/collect/UnmodifiableIterator;
at org.reflections.Reflections.expandSuperTypes(Reflections.java:380)
at org.reflections.Reflections.<init>(Reflections.java:126)
at org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.scanPluginPath(DelegatingClassLoader.java:221)
at org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.scanUrlsAndAddPlugins(DelegatingClassLoader.java:198)
at org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.registerPlugin(DelegatingClassLoader.java:190)
at org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.initLoaders(DelegatingClassLoader.java:150)
at org.apache.kafka.connect.runtime.isolation.Plugins.<init>(Plugins.java:47)
at org.apache.kafka.connect.cli.ConnectStandalone.main(ConnectStandalone.java:68)
我也试图在JAR文件移动到/app/kafka/libs directory
(默认CLASSPATH)甚至试图创建一个子目录/app/kafka/libs/connect_libs
和手动添加到我的CLASSPATH环境变量。
不知道我的下一步是什么,除了把卡尔卡和Elastic之间的Logstash。