问题标签 [confluent-schema-registry]

我使用 Confluent HDFS 连接器将数据从 Kafka 主题移动到 HDFS 日志文件。但是当我运行这些命令时：

./bin/connect-standalone etc/schema-registry/connect-avro-standalone.properties \ etc/kafka-connect-hdfs/quickstart-hdfs.properties

我正在采取跟随错误。我怎么解决这个问题。这是什么原因？

起因：org.apache.kafka.common.errors.SerializationException：反序列化 id -1 的 Avro 消息时出错 起因：org.apache.kafka.common.errors.SerializationException：未知魔术字节！[2017-06-03 13:44:41,895] 错误任务被杀死，直到手动重新启动才能恢复（org.apache.kafka.connect.runtime.WorkerTask:142）

我正在学习 Kafka，对我来说，使用 Avro 来使用带有 Schema 的 Kafka 主题是有意义的。

但是当谈到架构定义的放置位置时，我遗漏了一些东西：

• 如果我不使用模式注册表，但在我的项目中有 Avro 文件，我可以生成 Java 类并在发送消息时将其用作抽象层。这非常好，但现在我在多个项目中有这个文件的多个版本。我可以想象让它们保持同步会受到伤害。

• 如果我使用模式注册表，上面的问题就解决了。但是现在我看不到在生成消息时从模式定义中获利的方法：我需要手动生成 GenericRecord 对象以发送到 Kafka，而且我无法查看我生成的消息是否与模式匹配。

• 我也看不到使用模式来反序列化消费者端的消息的方法。

在序列化和反序列化消息时，有什么方法可以从模式定义中获利？

我找不到任何在两端都这样做的例子，特别是使用模式注册表。

我想知道我可以使用 Confluent Schema 注册表来生成（然后将其发送到 kafka）架构少 avro 记录吗？如果是的话，有人可以分享一些资源吗？我在 Confluent 网站和 Google 上找不到任何示例。

我有一个普通的分隔文件，我有一个单独的模式，目前我正在使用 Avro Generic Record 模式来序列化 Avro 记录并通过 Kafka 发送它。这样，模式仍然与记录相关联，这使得它更加庞大。我的逻辑是，如果我在从 kafka 发送记录时删除模式，我将能够获得更高的吞吐量。

即使服务器重新启动，我也希望能够保留具有固定 ID 的模式。

是否可以将模式保留在模式注册表中，以便在服务器崩溃后使它们具有相同的 ID？

否则，是否可以在模式注册服务器启动时硬编码具有固定 ID 的模式？

identityMapCapacityConfluent Schema Registry中的含义是什么CachedSchemaRegistryClient。根据文档，其声明如下：

我看到了几篇文章，其中初始化为int10，某处为 1000。所以我不确定它的确切含义以及我应该使用什么。

我在这里找到了一些代码https://docs.microsoft.com/en-us/azure/hdinsight/hdinsight-dotnet-avro-serialization#Scenario2与我需要的相反：

我想采用我创建的模型：

...并将此 C# 模型序列化为 JSON AVRO Schema 字符串。

原因：

我只想维护 C# 模型并使用 Confluent 的 Schema Registry 自动注册这些模型。要向架构注册表注册，架构需要采用 JSON AVRO 格式（就像Schema上面一样）。

我不希望同时定义 JSON 和 C# 模型。如果我必须维护一个，我宁愿拥有一个 C# 模型。

我是 Kafka 和 Avro 的菜鸟。所以我一直试图让生产者/消费者运行。到目前为止，我已经能够使用以下内容生成和使用简单的字节和字符串： 生产者的配置：

现在这一切都很好，当我尝试序列化 POJO 时问题就来了。因此，我能够使用 Avro 提供的实用程序从 POJO 获取 AvroSchema。对模式进行硬编码，然后尝试创建一个通用记录以通过 KafkaProducer 发送，生产者现在设置为：

这就是问题所在：当我使用 KafkaAvroSerializer 时，生产者没有出现，原因是： 缺少强制参数：schema.registry.url

我阅读了为什么需要这样做，以便我的消费者能够破译生产者发送给我的任何内容。但是模式不是已经嵌入到 AvroMessage 中了吗？如果有人可以分享一个使用 KafkaProducer 和 KafkaAvroSerializer 而无需指定 schema.registry.url 的工作示例，那就太好了

也非常感谢有关模式注册表实用程序的任何见解/资源。

谢谢！

我在使用响应式 kafka和avro4s生成 Avro 消息时遇到可重现的错误。一旦达到identityMapCapacity客户端 ( CachedSchemaRegistryClient) 的值，序列化就会失败

这是出乎意料的，因为所有消息都应该具有相同的模式——它们是相同案例类的序列化。

序列化器是 a KafkaAvroSerializer，用 a 实例化new CachedSchemaRegistryClient(settings.schemaRegistry, 1000)

生成GenericRecord：

架构是在代码深处创建的（io.confluent.kafka.serializers.AbstractKafkaAvroSerDe#getSchema，由 调用io.confluent.kafka.serializers.AbstractKafkaAvroSerializer#serializeImpl），我对其没有影响，所以我不知道如何修复泄漏。在我看来，这两个融合项目不能很好地协同工作。

我在这里、这里和这里发现的问题似乎并没有解决我的用例。

我目前的两种解决方法是：

• 不使用模式注册表——显然不是一个长期的解决方案
• 创建SchemaRegistryClient不依赖对象身份的自定义 - 可行但我想避免创建更多问题而不是重新实现

有没有办法根据消息/记录类型生成或缓存一致的模式并将其与我的设置一起使用？

我们正在编写一个自定义接收器连接器，用于将带有 avro 消息的主题内容写入 CEPH 存储。

为此，我们提供了 SinkRecords，它有一个 Kafka Connect 模式，它是我们的 avro 模式的映射版本。由于我们想将 avro 写入 CEPH，我们使用 connect API 方法将 Connect 模式转换回 Avro。为什么我们需要这样做？引入 Kafka Connect Schema 而不使用更常用的 Avro Schema 有什么好处？

仅供参考：我问这个是因为我们在 Avro 工会方面存在一些问题。他们到 Kafka Connect Schema 的映射仍然存在一些问题，例如https://github.com/confluentinc/schema-registry/commit/31648f0d34b10c1b36e8ec6f4c1236ed3fe86495#diff-0a8d4f17f8d4a68f2f0d2dcd9211df84

设置：-我们在 AWS ELB 后面有 3 个 Schema 注册表实例。如何更改 schema_registry.properties 文件以在集群模式下设置模式注册表？

我们正在使用 ELB 端点调用模式注册表。