Flinkkafka消费者/制作人

Question

大家好，我正在努力将一个简单的avro模式与模式注册表一起序列化。

设置：

• 两个用java编写的Flink jobs（一个消费者，一个生产者）

目标：生产者应该发送一条用ConfluentRegistryAvroSerializationSchema序列化的消息，其中包括更新和验证模式。

然后，使用者应将消息反序列化为具有接收到的模式的对象。使用ConfluentRegistryAvroDeserializationSchema。

到目前为止还不错：如果我将架构注册表上的主题配置为FORWARD-兼容，生产者将正确的avro架构写入注册表，但它以错误结束（即使我完全永久地删除主题第一）：

Failed to send data to Kafka: Schema being registered is incompatible with an earlier schema for subject "my.awesome.MyExampleEntity-value"

已成功编写架构：

{
        "subject": "my.awesome.MyExampleEntity-value",
        "version": 1,
        "id": 100028,
        "schema": "{\"type\":\"record\",\"name\":\"input\",\"namespace\":\"my.awesome.MyExampleEntity\",\"fields\":[{\"name\":\"serialNumber\",\"type\":\"string\"},{\"name\":\"editingDate\",\"type\":\"int\",\"logicalType\":\"date\"}]}"
}

接下来，我可以尝试将可兼容性设置为NONE

如果我这样做，我可以在Kafka上生成数据，但是：模式注册表有一个新版本的我的模式，如下所示：

{
        "subject": "my.awesome.MyExampleEntity-value",
        "version": 2,
        "id": 100031,
        "schema": "\"bytes\""
}

现在，我可以生成数据，但消费者无法反序列化此模式，并发出以下错误：

Caused by: org.apache.avro.AvroTypeException: Found bytes, expecting my.awesome.MyExampleEntity
...

我目前不确定问题到底出在哪里。即使我完全永久地删除主题（包括模式），我的生产者也应该从头开始正常工作，用模式注册一个完整的“新”主题。另一方面，如果我将兼容性设置为“NONE”，模式交换无论如何都应该通过注册一个可以被消费者读取的模式来工作。

有人能帮我吗？

Answer 1

序列化的整个问题是关于kafka配置中以下标志的使用：

"schema.registry.url"
"key.serializer"
"key.deserializer"
"value.serializer"
"value.deserializer"

在flink中设置这些标志，即使它们在逻辑上是正确的，也会导致不可调试的模式验证和序列化混乱。省略了所有这些标志，效果很好。注册表url只需在ConfluentRegistryAvro（De）serializationSchema中设置。

Answer 2

根据最新的合流文档NONE：模式兼容性检查被禁用文档：