使用特定 ID 将架构添加到架构注册表

Question

我们使用 Confluence SchemaRegistry 和 KafkaStreams 已经一年多了，一切都运行良好；直到昨天。

在 UAT 环境中，我们似乎删除了一个架构主题，并且我们的一个应用程序开始故障转移并显示以下消息

[错误] LogAndFailExceptionHandler - 反序列化期间捕获异常，taskId：0_13，主题：TOPIC_NAME，分区：13，偏移量：0 org.apache.kafka.common.errors.SerializationException：检索 id 1531 的 Avro 架构时出错

我检查了架构注册表，发现主题丢失了，并使用curl查询错误中列出的id 1531，例如：

curl -X GET http://SchemaRegistryHost:8081/schemas/ids/1531

并回来了：

{"error_code":40403,"message":"Schema not found"}

我天真地只是尝试再次注册架构，没有考虑它，它起作用了，但是注册架构的 id 与之前的 1531 ID 不同。

我需要将架构注册到 ID 1531，因为主题中的现有消息已在魔术字节中包含该 Id 1531。

我在https://docs.confluence.io/current/schema-registry/docs/develop/api.html检查了 API 文档，但没有看到任何用于为模式设置给定 Id 的内容。

无论如何，是否可以使用架构注册表将架构强制指定为特定 ID？

我知道一些备份解决方案，但我现在正在寻找一种修复方法，希望能够防止数据丢失或采取特殊措施来修复主题数据。

Answer 1

现在有可能了！

做这个

curl -X POST -H "Content-Type: application/json" \
--data '{"schemaType": "AVRO", "version":1, "id":24, "schema":"{\"type\":\"record\",\"name\":\"value_a1\",\"namespace\":\"com.mycorp.mynamespace\",\"fields\":[{\"name\":\"field1\",\"type\":\"string\"}]}" }' \
http://localhost:8081/subjects/my-cool-subject/versions

请参阅： https: //docs.confluence.io/platform/current/schema-registry/installation/migrate.html#migrate-an-individual-schema-to-an-already-populated-sr

Answer 2

无论如何，是否可以使用架构注册表将架构强制指定为特定 ID？

那没有。

顺便说一句，ID 1531 实际上并没有“消失”，它只是在注册表中标记_schemas为已删除（使用主题查看它）。

据我所知，当您使用 KafkaAvroDeserializer 时，确实无法解决该错误。您必须使用 ByteArrayDeserializer，然后使用架构注册表客户端“修复”或“查找”正确的 ID，然后反序列化消息的其余部分。

另一种选择是重置您的消费者组，以便完全跳过这些消息，或者设置异常处理。使用 Kafka 的 Streams API 处理不良消息