我是新使用Kafka和Kafka连接器。我一直在尝试使用elastic search作为接收器,使用kafka连接器从我的应用程序流式传输数据。我可以看到Kafka中的信息,但我的连接器不断抛出以下错误:
ERROR WorkerSinkTask{id=elasticsearch-sink-0} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask:179)
org.apache.kafka.connect.errors.ConnectException: Couldn't start ElasticsearchSinkTask due to connection error:
at io.confluent.connect.elasticsearch.jest.JestElasticsearchClient.<init>(JestElasticsearchClient.java:159)
at io.confluent.connect.elasticsearch.jest.JestElasticsearchClient.<init>(JestElasticsearchClient.java:142)
at io.confluent.connect.elasticsearch.ElasticsearchSinkTask.start(ElasticsearchSinkTask.java:122)
at io.confluent.connect.elasticsearch.ElasticsearchSinkTask.start(ElasticsearchSinkTask.java:51)
at org.apache.kafka.connect.runtime.WorkerSinkTask.initializeAndStart(WorkerSinkTask.java:300)
at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:189)
at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:177)
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:227)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Caused by: io.searchbox.client.config.exception.CouldNotConnectException: Could not connect to http://elasticsearch:9200
at io.searchbox.client.http.JestHttpClient.execute(JestHttpClient.java:73)
at io.searchbox.client.http.JestHttpClient.execute(JestHttpClient.java:63)
at io.confluent.connect.elasticsearch.jest.JestElasticsearchClient.getServerVersion(JestElasticsearchClient.java:247)
at io.confluent.connect.elasticsearch.jest.JestElasticsearchClient.<init>(JestElasticsearchClient.java:151)
... 12 more
Caused by: org.apache.http.conn.HttpHostConnectException: Connect to elasticsearch:9200 [elasticsearch/172.20.0.7] failed: Connection refused (Connection refused)
at org.apache.http.impl.conn.DefaultHttpClientConnectionOperator.connect(DefaultHttpClientConnectionOperator.java:159)
at org.apache.http.impl.conn.PoolingHttpClientConnectionManager.connect(PoolingHttpClientConnectionManager.java:359)
at org.apache.http.impl.execchain.MainClientExec.establishRoute(MainClientExec.java:381)
at org.apache.http.impl.execchain.MainClientExec.execute(MainClientExec.java:237)
at org.apache.http.impl.execchain.ProtocolExec.execute(ProtocolExec.java:185)
at org.apache.http.impl.execchain.RetryExec.execute(RetryExec.java:89)
at org.apache.http.impl.execchain.RedirectExec.execute(RedirectExec.java:111)
at org.apache.http.impl.client.InternalHttpClient.doExecute(InternalHttpClient.java:185)
at org.apache.http.impl.client.CloseableHttpClient.execute(CloseableHttpClient.java:83)
at org.apache.http.impl.client.CloseableHttpClient.execute(CloseableHttpClient.java:108)
at io.searchbox.client.http.JestHttpClient.executeRequest(JestHttpClient.java:136)
at io.searchbox.client.http.JestHttpClient.execute(JestHttpClient.java:70)
... 15 more
Caused by: java.net.ConnectException: Connection refused (Connection refused)
at java.net.PlainSocketImpl.socketConnect(Native Method)
at java.net.AbstractPlainSocketImpl.doConnect(AbstractPlainSocketImpl.java:350)
at java.net.AbstractPlainSocketImpl.connectToAddress(AbstractPlainSocketImpl.java:206)
at java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:188)
at java.net.SocksSocketImpl.connect(SocksSocketImpl.java:392)
at java.net.Socket.connect(Socket.java:589)
at org.apache.http.conn.socket.PlainConnectionSocketFactory.connectSocket(PlainConnectionSocketFactory.java:75)
at org.apache.http.impl.conn.DefaultHttpClientConnectionOperator.connect(DefaultHttpClientConnectionOperator.java:142)
... 26 more
[2020-09-22 08:16:32,656] ERROR WorkerSinkTask{id=elasticsearch-sink-0} Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerTask:180)
我已经对接了所有东西,我有4个容器,一个用于我的应用程序,一个用于elasticsearch,一个用于kafka,一个用于连接器,如下所示:
zookeeper:
container_name: zookeeper
image: wurstmeister/zookeeper:latest
env_file:
- ".env"
ports:
- 2181:2181
networks:
- "main_net"
kafka:
container_name: kafka
image: wurstmeister/kafka:2.11-1.0.2
env_file:
- ".env"
depends_on:
- zookeeper
links:
- zookeeper
ports:
- 9092:9092
- 9094:9094
networks:
- "main_net"
connector_standalone:
container_name: container_standalone
build:
context: kafka/
dockerfile: Dockerfile
depends_on:
- kafka
ports:
- 8083:8083
networks:
- "main_net"
elasticsearch:
container_name: elasticsearch
build:
context: elasticsearch/
ports:
- "9200:9200"
- "9300:9300"
env_file:
- ".env"
environment:
discovery.type: single-node
networks:
- "main_net"
website:
container_name: "application"
build: "./"
command: >
volumes:
- "./application:/app"
ports:
- "8000:8000"
networks:
- "main_net"
据我所知,我已经给出了连接url,基于它应该连接到的主机,即elasticsearch容器。但我很困惑,基于错误什么是错的。下面是我的配置文件:
name=elasticsearch-sink
connector.class=io.confluent.connect.elasticsearch.ElasticsearchSinkConnector
tasks.max=1
topics=vehicle
topic.index=vehicles
connection.url=http://elasticsearch:9200/
connection.user=elastic
connection.password=changeme
type.name=log
key.ignore=true
schema.ignore=true
1条答案
按热度按时间dphi5xsq1#
因为这个它失败了
因此,要么elasticsearch尚未启动,要么它正在启动,但不接受连接。
在创建kafka connect连接器之前,请通过运行以下命令确保kafka connect worker提供elasticsearch
你应该把这样的东西拿回来
fwiw您可以在这里找到使用docker compose的kafka->elasticsearch的工作示例(?)