如何在不重启容器的情况下更新Kafka Connect插件

Kafka Connect 2.3.0及以上版本自带插件热重载能力，不需要改源码、装第三方工具就能实现你要的效果，你现在必须重启容器是因为默认配置关闭了运行期插件扫描。

核心配置修改 #

启动Kafka Connect容器时，调整以下配置参数即可开启热加载：

• 保持plugin.path配置指向你绑定挂载的插件目录，和你当前的挂载路径一致即可，比如常用的路径是/opt/kafka/plugins
• 关闭启动时单次扫描限制：设置plugin.discovery.only.startup=false，这个参数默认值为true，也就是仅在服务启动阶段扫描一次插件目录，改成false后会在服务运行期间定期扫描
• 配置扫描周期：设置plugin.scan.interval.ms=5000，单位为毫秒，开发调试阶段设为5000（即每5秒扫描一次）就足够，生产环境可以适当调长或者关闭热扫描。

如果你用的是Confluent官方提供的Kafka Connect镜像，不需要进容器手动改配置文件，直接传CONNECT_前缀的环境变量就能自动映射配置，docker-compose配置片段参考：

kafka-connect:
  image: confluentinc/cp-kafka-connect:7.5.0
  ports:
    - "8083:8083"
  environment:
    CONNECT_BOOTSTRAP_SERVERS: kafka:9092
    CONNECT_REST_ADVERTISED_HOST_NAME: kafka-connect
    CONNECT_GROUP_ID: connect-cluster
    CONNECT_CONFIG_STORAGE_TOPIC: connect-configs
    CONNECT_OFFSET_STORAGE_TOPIC: connect-offsets
    CONNECT_STATUS_STORAGE_TOPIC: connect-statuss
    CONNECT_PLUGIN_PATH: "/opt/kafka/plugins,/usr/share/java"
    # 热重载核心配置
    CONNECT_PLUGIN_DISCOVERY_ONLY_STARTUP: "false"
    CONNECT_PLUGIN_SCAN_INTERVAL_MS: "5000"
  volumes:
    - ./your-local-plugin-dir:/opt/kafka/plugins

替换插件后的操作 #

配置生效后，你编译完新的connector jar包，替换挂载目录里的旧文件后，不需要重启容器，等一个扫描周期（最多等你配置的间隔时长）后，执行对应操作即可：

• 已经创建过实例的插件：调用Connect REST接口重启对应connector即可生效，命令参考：

  curl -X POST http://<你的connect服务地址>:8083/connectors/<你的connector名称>/restart?includeTasks=true&onlyFailed=false
  

• 全新开发的、没创建过实例的插件：等扫描周期过了之后，直接正常调用REST接口创建connector实例就行，服务能直接识别到新插件。

我本地开发自定义connector的时候一直用这套配置，之前每次改完代码要等30多秒容器重启，现在编译完丢包等5秒调个重启接口就能测，调试效率提升很明显。

注意事项 #

• 不要把新旧版本的插件jar同时放在插件目录里，很容易引发类加载冲突，替换的时候建议直接删掉对应插件的整个子目录，再把新编译好的、带全依赖的插件文件夹放进去
• 插件不要直接丢在plugin.path的根目录，要给每个插件建单独的子目录，比如/opt/kafka/plugins/my-custom-source-connector/下放所有相关jar，不然扫描逻辑可能识别不到插件
• 替换正在运行的connector对应的插件时，建议先暂停connector再替换文件，避免扫描加载过程中处理请求抛类找不到的异常
• 如果你用的Kafka Connect版本低于2.3.0，没有内置热扫描能力，要么升级版本，要么用JRebel这类字节码热替换工具做调试，不过还是升级版本用官方能力最稳定，没有额外依赖。

热重载逻辑只会扫描识别新插件、新版本插件，不会自动重启正在运行的connector任务，必须手动调用重启接口才会加载新的类逻辑。

内容的提问来源于stack exchange，提问作者Yannis Yahi