Kafka使用者模式注册表动态主题名称

Question

我有多个主题A、B和C，我正在使用Kafka Streams将它们扇形成主题X。主题A、B和C使用默认主题名称策略在模式注册表中注册为主题。流式传输非常愚蠢，因为它只是将消息扇形，而不确保它们符合注册表中的模式，但它会在消息中添加一个ORIGINAL_TOPIC_NAME标头以指示它来自主题A、B或C。

然后，我让Kafka消费者从主题X中消费。该主题未在模式注册表中注册。Kafka消费者是我使用Kafkaavroderialiser和schema.registry的地方。使用注册表消费的url。我计划让消费者在这里对注册表进行检查，但使用原始的_TOPIC\u NAME头作为主题。但是，我不确定是否可以控制消费者，使其使用Kafka头解析主题名称，因为主题名称策略是您在初始化时输入消费者的内容。

有什么想法吗？

Answer 1

策略传递给反序列化器

反序列化函数只能访问键或值的传入字节数组以及主题名，而不能访问头。

可以设置头反序列化器，但它不在键/值反序列化器的调用路径中使用。

解决方法是对键/值使用ByteArrayDeserializer，可以选择手动提取头，使用如下逻辑获取模式ID

ByteBuffer bb = ByteBuffer.wrap(value); // or key
if (bb.get() != 0x0) {
    // not a valid encoded Schema Registry record 
}
int schemaId = bb.getInt();
// Completely ignores the subject 
Schema schema = schemaRegistryClient.getId(schemaId);
// TODO: continue to deserialize the key/value byte array