如何彻底移除Kafka Connect连接器及实现容器停服自动删除？

一、彻底移除连接器后同名重建是否可行？ #

完全可行，但需在删除连接器基础上，额外清理关联的偏移量与状态数据，具体操作如下：

• 删除连接器实例：通过Connect REST API执行删除命令，终止连接器运行并移除配置：

  curl -X DELETE http://<connect-host>:8083/connectors/<your-connector-name>
  

  此操作会向Connect配置主题发送墓碑消息，但偏移量、状态数据仍保留在对应存储主题中。
• 清理偏移量数据：偏移量存储在offset.storage.topic指定的Kafka主题中。先定位目标连接器的偏移量key：

  kafka-console-consumer.sh --bootstrap-server <broker-address> \
    --topic <offset-storage-topic> \
    --property print.key=true \
    --property key.deserializer=org.apache.kafka.common.serialization.StringDeserializer \
    --property value.deserializer=org.apache.kafka.common.serialization.StringDeserializer | grep <your-connector-name>
  

  再发送墓碑消息删除对应记录：

  echo "<target-key>,null" | kafka-console-producer.sh --bootstrap-server <broker-address> \
    --topic <offset-storage-topic> \
    --property parse.key=true \
    --property key.separator=, \
    --property key.serializer=org.apache.kafka.common.serialization.StringSerializer \
    --property value.serializer=org.apache.kafka.common.serialization.StringSerializer
  

• 清理状态数据：状态数据存储在status.storage.topic指定的主题中，操作逻辑与偏移量清理一致，找到对应记录并发送墓碑消息删除。
  完成所有步骤后，即可用同名重新创建连接器，不会因残留数据产生冲突。

二、让Kafka Connect容器停止时自动删除所有连接器的方法 #

通过容器生命周期钩子+自定义脚本可实现，具体方案如下：

1. 编写清理脚本：在容器内创建cleanup-connectors.sh脚本，内容如下：

   #!/bin/bash
   # 捕获停止信号，执行连接器清理
   trap 'echo "Starting connector cleanup..."; \
         CONNECTORS=$(curl -s http://localhost:8083/connectors | jq -r ".[]"); \
         for CONNECTOR in $CONNECTORS; do \
             echo "Deleting connector: $CONNECTOR"; \
             curl -X DELETE http://localhost:8083/connectors/$CONNECTOR; \
         done; \
         exit 0' SIGTERM SIGINT
   
   # 启动Kafka Connect（替换为你的容器启动命令）
   /etc/confluent/docker/run &
   wait $!
   

2. 配置容器：将脚本设置为容器入口点，确保捕获停止信号触发清理：
   • Dockerfile示例：

     FROM confluentinc/cp-kafka-connect:latest
     COPY cleanup-connectors.sh /usr/local/bin/
     RUN chmod +x /usr/local/bin/cleanup-connectors.sh
     ENTRYPOINT ["cleanup-connectors.sh"]
     

   • docker-compose.yml示例：

     services:
       kafka-connect:
         image: confluentinc/cp-kafka-connect:latest
         volumes:
           - ./cleanup-connectors.sh:/usr/local/bin/cleanup-connectors.sh
         entrypoint: ["cleanup-connectors.sh"]
         stop_signal: SIGTERM
     

3. 注意：若为分布式Connect集群，此操作会删除集群内所有连接器，需确认业务场景允许；脚本中需确保能访问Connect的REST API，端口非8083时需对应调整。

内容的提问来源于stack exchange，提问作者Omri. B