我使用的是汇合平台3.2。在3台不同的ec2机器上运行3名工人。
我有一个连接器(debezium/mysql源代码),我删除了它,几分钟后重新启动。但由于以下错误,我无法成功启动连接器。连接器处于故障状态。我不得不重新启动工人来解决这个问题。
需要知道这是否是缓存的问题吗?如何在不重新启动工人的情况下解决此问题。感谢您的支持。
{
"name": "debezium-connector",
"connector": {
"state": "RUNNING",
"worker_id": "xx.xx.xx.xxx:8083"
},
"tasks": [
{
"state": "FAILED",
"trace": "org.apache.kafka.connect.errors.ConnectException: Task already exists in this worker: debezium-connector-0\n\tat org.apache.kafka.connect.runtime.Worker.startTask(Worker.java:308)\n\tat org.apache.kafka.connect.runtime.distributed.DistributedHerder.startTask(DistributedHerder.java:834)\n\tat org.apache.kafka.connect.runtime.distributed.DistributedHerder.access$1500(DistributedHerder.java:101)\n\tat org.apache.kafka.connect.runtime.distributed.DistributedHerder$13.call(DistributedHerder.java:848)\n\tat org.apache.kafka.connect.runtime.distributed.DistributedHerder$13.call(DistributedHerder.java:844)\n\tat java.util.concurrent.FutureTask.run(FutureTask.java:266)\n\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)\n\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)\n\tat java.lang.Thread.run(Thread.java:745)\n",
"id": 0,
"worker_id": "xx.xx.xx.xxx:8083"
}
]
}
1条答案
按热度按时间zazmityj1#
嗯,我也有同样的错误,然后我发现Kafka服务器之一的磁盘空间用完了,因此Kafka集群不能正常工作。我不知道这里的所有细节,但我希望connect正在kafka中存储一些关于连接器和任务的信息,如果它没有正确响应,kafka可能仍然有关于旧任务的信息。
分享以防对其他人有帮助。
编辑:
我还注意到,这个问题时常发生在我的kafka节点上,使整个集群处于不可用状态。重新启动有问题的节点可以解决这些问题。