Kafka Connect创建Oracle连接器无Worker生成，需重启才可正常工作

问题描述 #

通过Put/Push请求创建Debezium Oracle连接器后，连接器显示创建成功，但后续无任何操作，直至API返回500，日志中出现超时信息，且未生成Worker任务及相关主题。重启连接器后，任务及主题可正常生成。该问题仅在特定场景下触发。

连接器配置 #

{"connector.class": "io.debezium.connector.oracle.OracleConnector","transforms.convert_scns.type": "org.apache.kafka.connect.transforms.Cast$Value","transforms.unwrap.delete.handling.mode": "rewrite","topic.creation.default.partitions": "5","transforms": "unwrap,convertTS,convert_scns,reroute","transforms.reroute.type": "io.debezium.transforms.ByLogicalTableRouter","transforms.convertTS.format": "EEE MMM dd HH:mm:ss zzz yyyy","transforms.convertTS.target.type": "Timestamp","schema.history.internal.store.only.captured.tables.ddl": "true","topic.prefix": "topic_name","decimal.handling.mode": "double","schema.history.internal.kafka.topic": "schema-changes","transforms.unwrap.drop.tombstones": "false","topic.creation.default.replication.factor": "2","transforms.convertTS.field": "__source_ts_ms","transforms.reroute.topic.regex": "topic_name\\.([^.]+)\\.table_name","transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState","value.converter": "io.confluent.connect.avro.AvroConverter","transforms.convertTS.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value","key.converter": "io.confluent.connect.avro.AvroConverter","database.user": "","database.dbname": "","transforms.reroute.topic.replacement": "topic_name","time.precision.mode": "connect","schema.history.internal.kafka.bootstrap.servers": "ip:9092,ip:9092,ip:9092","database.port": "","value.converter.schema.registry.url": "http://ip:8085","database.hostname": "","database.password": "","name": "name","numeric.mapping": "best_fit","transforms.unwrap.add.fields": "op,source.ts_ms,source.schema,source.scn,source.commit_scn,source.db","table.include.list": "table_name","key.converter.schema.registry.url": "http://ip:8085","snapshot.mode": "initial","transforms.convert_scns.spec": "__source_commit_scn:int64,__source_scn:int64,__deleted:boolean"}

连接器创建日志 #

Instantiated connector test_connector with version 2.1.3.Final of type class io.debezium.connector.oracle.OracleConnector (org.apache.kafka.connect.runtime.Worker:315)
[2023-04-17 17:37:10,376] INFO [test_connector|worker] Finished creating connector test_connector (org.apache.kafka.connect.runtime.Worker:336)
[2023-04-17 17:37:10,377] DEBUG [test_connector|worker] WorkerConnector{id=test_connector} Initializing connector test_connector (org.apache.kafka.connect.runtime.WorkerConnector:166)
[2023-04-17 17:37:10,377] DEBUG [test_connector|worker] WorkerConnector{id=test_connector} Transition connector to STARTED (org.apache.kafka.connect.runtime.WorkerConnector:355)

触发场景 #

• JSONConverter：可正常工作
• JSONConverter搭配Schema Registry：可正常工作
• AvroConverter搭配Schema Registry：出现挂起，需重启连接器才可正常运行

排查思路 #

• 聚焦AvroConverter与Schema Registry的交互逻辑：问题仅出现在该组合，优先排查两者初始化阶段的协作流程
• 验证Schema Registry的可用性：检查连接器启动时是否能快速访问Schema Registry，是否存在网络延迟、连接超时或权限限制
• 开启详细日志：将Kafka Connect日志级别调整为TRACE，查看连接器初始化AvroConverter、注册Schema时的具体执行步骤，定位是否存在阻塞点或未捕获异常
• 检查Kafka Connect超时配置：确认连接器创建请求的超时参数是否过短，导致AvroConverter未完成初始化就触发API超时
• 验证Schema注册逻辑：检查连接器生成的初始表Schema是否符合Schema Registry的格式要求，是否存在兼容性问题

解决方案 #

• 调整Kafka Connect超时参数：设置connector.client.config.override.policy=All，并增大request.timeout.ms、connection.timeout.ms等参数，给AvroConverter足够的初始化时间
• 预注册Schema：提前将目标表的Schema手动注册到Schema Registry，避免连接器启动时动态注册的阻塞
• 升级组件版本：检查Debezium 2.1.3.Final与AvroConverter、Schema Registry的版本兼容性，升级到匹配的稳定版本
• 简化初始配置：移除非必要的Transform（如reroute、convertTS），先验证基础AvroConverter配置的可用性，再逐步添加Transform
• 排查网络与权限：确保Kafka Connect节点能正常访问Schema Registry，无防火墙或代理拦截，且Schema Registry的访问权限配置正确

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