Kafka Connect连接器如何确定对应主题使用的Schema？

核心匹配规则（默认场景） #

Confluent Schema Registry和对应的Connect转换器默认遵循主题关联的Subject命名策略，无需手动绑定Topic与Schema：

• 当生产者使用JsonSchemaSerializer序列化消息时，会自动将Schema注册到以{topic}-value（消息体）或{topic}-key（消息键）命名的Subject下；
• Sink Connector的JsonSchemaConverter会根据当前处理的Topic名称，自动拼接出对应的Subject（比如处理order-updates Topic时，会查找order-updates-value Subject），然后从Registry拉取对应Schema完成反序列化。

必须满足的前提条件 #

1. 生产者端配置正确
   生产者必须使用Confluent提供的Schema Registry序列化器（如io.confluent.kafka.serializers.json.JsonSchemaSerializer），且未自定义Subject命名策略。若生产者手动指定了非默认Subject，后续需在Connector端同步配置。

2. Sink Connector转换器配置正确
   连接器必须使用支持Schema Registry的转换器，不能用普通的JsonConverter。关键配置示例：

   value.converter=io.confluent.connect.json.JsonSchemaConverter
   value.converter.schema.registry.url=http://your-schema-registry:8081
   # 若需处理消息键，同理配置key.converter
   key.converter=io.confluent.connect.json.JsonSchemaConverter
   key.converter.schema.registry.url=http://your-schema-registry:8081
   

自定义Subject命名策略的处理 #

如果生产者使用了自定义Subject命名策略（比如TopicRecordNameStrategy，按Topic+记录类名生成Subject），则需要在Sink Connector的转换器配置中指定相同策略：

value.converter.subject.name.strategy=io.confluent.kafka.serializers.subject.TopicRecordNameStrategy

这样连接器会按照相同规则生成Subject名称，匹配到正确的Schema。

验证匹配有效性的方法 #

• 调用Schema Registry的REST接口GET /subjects，确认对应Topic的{topic}-value/{topic}-key Subject已存在；
• 查看Connector运行日志，若出现Schema not found类错误，会明确提示缺失的Subject名称，可据此排查注册或配置问题。

内容的提问来源于stack exchange，提问作者Kjell van Straaten