Confluent 4.1.0中topic_out-key主题缺失引发Schema查询报错咨询

1. 为什么topic_out-key Schema主题没有自动创建？ #

在Confluent 4.1.0版本中，KTable.toStream()输出流时不会自动为消息的Key注册Schema，除非你在流处理拓扑中显式配置了Key的Avro序列化器并正确关联Schema Registry。具体可能的原因包括：

• 你的流处理代码仅为消息Value配置了Avro Serde（比如GenericAvroSerde或SpecificAvroSerde），但Key使用默认Serde（如StringSerde），或未为Key指定Avro相关序列化配置。即便实际输出的Key是Avro格式，Producer也不会主动触发Key Schema的注册流程。
• 执行KTable关联操作后，输出流的Key类型未被显式声明为Avro兼容类型（如GenericRecord或自定义Avro生成类），导致Producer无法识别需要为Key注册Schema。
• Producer配置缺失Key序列化的关键参数：未设置key.serializer=io.confluent.kafka.serializers.KafkaAvroSerializer，也未配置schema.registry.url指向你的Schema Registry地址，直接导致Key Schema的自动注册逻辑未被触发。

另外，从你提供的Schema Registry日志来看，有尝试删除topic_out-key的请求但返回404，这说明之前可能有清理操作，但核心问题还是初始生成流时未成功注册Key的Schema。

2. 为什么必须存在Key对应的Schema主题？ #

HDFS Sink Connector消费topic_out时，若你配置了key.converter=io.confluent.connect.avro.AvroConverter（这是Avro场景下的常见配置），会尝试同时反序列化消息的Key和Value。当Connector读取到Avro格式的Key时，会根据Key中携带的Schema ID（日志里的ID 5）去Schema Registry查找对应的Schema定义，而这个Schema正是存储在topic_out-key主题下的。

如果topic_out-key主题不存在，Schema Registry无法找到对应ID的Schema，就会抛出Subject not found的404错误，导致Connector无法完成反序列化，进而任务失败。简单来说：只要消息Key使用Avro序列化，消费端（如Sink Connector）就必须能从Schema Registry获取对应的Key Schema，而这个Schema的存储依赖于{topic}-key主题的存在。

临时修复建议（针对你的场景） #

你可以手动注册Key的Schema到topic_out-key主题来快速解决问题：

1. 通过kafka-avro-console-consumer获取一条topic_out的Key数据，导出其Schema（可使用--property print.schema.ids=true查看Key的Schema ID，或反序列化后提取Schema内容）。
2. 使用Schema Registry API手动注册：

   curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
   --data '{"schema": "YOUR_KEY_SCHEMA_JSON"}' \
   http://localhost:8081/subjects/topic_out-key/versions
   

3. 重启HDFS Sink Connector，验证是否能正常消费。

长期解决方案 #

在流处理代码中，显式为输出流配置Key的Avro Serde，并确保Producer配置包含Key序列化的必要参数：

// 配置Key和Value的Avro Serde
final Serde<GenericRecord> keyAvroSerde = new GenericAvroSerde();
final Serde<GenericRecord> valueAvroSerde = new GenericAvroSerde();

final Map<String, String> serdeConfig = Collections.singletonMap(
  AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081"
);

keyAvroSerde.configure(serdeConfig, true); // true表示这是Key的Serde
valueAvroSerde.configure(serdeConfig, false); // false表示这是Value的Serde

// 输出流时指定Key和Value的Serde
ktable.toStream().to("topic_out", Produced.with(keyAvroSerde, valueAvroSerde));

这样，流处理程序运行时，Producer会自动将Key的Schema注册到topic_out-key主题，后续Connector就能正常获取Schema了。

内容的提问来源于stack exchange，提问作者K.Ketan