自定义Kafka Connector任务配置创建超时问题求助

可以通过调整Kafka Connect的相关配置参数，延长任务配置生成的等待超时时间，避免Connector被不必要地重新调度，进而防止任务数量翻倍的问题。具体调整方案如下：

关键配置参数调整 #

• connector.client.config.override.policy
  首先需要开启允许覆盖Connector级别的配置参数，默认值为None，无法自定义超时参数。将其设置为All（允许所有配置覆盖）或Principal（基于权限控制的覆盖），这样才能在Connector层面或全局层面调整超时参数。

• task.config.generation.timeout.ms
  这个参数控制等待任务配置生成的超时时间，默认值通常为5分钟（300000毫秒），刚好低于你的6分钟计算耗时。直接将其设置为大于6分钟的值，比如360000（6分钟）或420000（7分钟），确保框架有足够时间等待taskConfigs方法完成计算。

• connect.connector.rebalance.timeout.ms
  该参数控制集群重新平衡的超时时间，当Connector调度时，框架会等待重新平衡完成。如果任务配置生成过慢，可能会触发重新平衡超时导致重调度，建议将此值也适当调大，比如设置为420000（7分钟），给整个调度流程预留足够缓冲时间。

配置方式 #

1. 全局配置（所有Connector生效）
   在Kafka Connect worker的配置文件（如connect-distributed.properties）中添加以下配置：

   connector.client.config.override.policy=All
   task.config.generation.timeout.ms=360000
   connect.connector.rebalance.timeout.ms=420000
   

2. 单独配置（仅当前自定义Connector生效）
   如果不想影响其他Connector，可以在创建/更新Connector的请求中指定这些参数：

   {
     "name": "your-custom-connector",
     "config": {
       "connector.class": "com.your.package.CustomConnector",
       "task.config.generation.timeout.ms": "360000",
       "connect.connector.rebalance.timeout.ms": "420000",
       // 其他Connector专属配置...
     }
   }
   

注意事项 #

• 调整超时时间后，要确保worker节点有足够的CPU、内存资源支撑长时间的任务配置计算，避免因资源不足导致计算耗时进一步增加或worker崩溃。
• 全局调整会影响所有Connector的调度行为，若集群内有其他对响应速度要求高的Connector，建议优先使用Connector单独配置的方式。

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