问题标签 [confluent-schema-registry]

我正在尝试使用 FlinkKafkaProducer10 从 Flink 的模式注册表写入 Confluent kafka。产生以下错误。我创建了自定义模式序列化程序，请参阅 ConfluentAvroSerializationSchema 类。代码编译但产生运行时错误。产生错误的完整代码示例在https://github.com/dmiljkovic/test-flink-schema-registry上。所有资源都是模拟的，kafka 集群和模式注册表。代码实际上是测试用例。

When I used Schema registry server, I found I could not change schema in the same topic anymore. but without schema registry each record has it's own schema inside, so it's easy to change schema anytime.

So how can I change the schema when I'm using schema registry server.

有没有人使用过 spring-kafka 2.0.0.Release 并创建了一个使用融合模式注册表作为反序列化消息源的消费者？如果是这样，你能给我举个例子吗？

我要解决的问题是我的 kafka 连接平台上有一个 Debezium CDC 连接器，它可以在事件发生时从 MongoDB 流式传输事件。我必须拦截这些事件转换并重新传输。要了解该事件，我必须反序列化有效负载。我目前卡在这一步。

在尝试使用 Kafka Streams 流式传输 Avro 数据时，我遇到了这个错误：

Exception in thread "StreamThread-1" org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id -1 Caused by: org.apache.kafka.common.errors.SerializationException: Unknown magic byte!

尽管我在邮件列表中找到了几个关于它的旧线程，但没有一个解决方案说明了该问题。所以希望我能在这里找到解决方案。

我的设置如下所示：

我已经尝试将 设置KEY_SERDE为与 相同VALUE_SERDE，但即使这在邮件列表中被“标记”为解决方案，它在我的情况下也不起作用。

我正在GenericData.Record使用我的 Schema 生成如下：

当我启动调试模式并检查生成的记录时，一切看起来都很好，记录中有数据并且映射正确。

我像这样流式传输 KStream（我之前使用过分支）：

splitTopics.get(0).to(s"${destTopic}_Testing")

我是GenericData.Record用来记录的。这可能是与 结合使用的问题GenericAvroSerde吗？

我正在寻找安装融合模式注册表的选项，是否可以单独下载和安装注册表并使其与现有的 kafka 设置一起使用？

谢谢

我刚刚完成了本教程以使用 Kafka 和 Schema Registry：http ://cloudurable.com/blog/kafka-avro-schema-registry/index.html 我还玩过 Conlfuent Platform：https ://docs.confluent.io/当前/安装/installing_cp.html

一切正常，直到我重新启动我的虚拟机（VMBOX）：重新启动 后所有模式/主题都已删除（或消失）。

我读了那个 Schema Registry 来不存储自己的数据，而是使用 Kafka 来做到这一点。当然，由于我目前只在笔记本电脑上工作，因此在机器重启期间 Kafka 也关闭了。

这是正常行为吗，我们是否必须期望在重新启动时一直重新存储所有模式？？？（->也许是最后一个版本！）

有人对此有良好的最佳实践吗？

如何管理模式的持久性以避免这个问题？

环境：Ubuntu 16...，Kafka 2.11.1.0.0，Confluent Platform 4.0

非常感谢

注意：我已经阅读了讨论有关保留模式 ID 的主题，但是我没有恢复任何模式，这不是 Ids 的问题：Confluent Schema Registry Persistence

我们计划通过 Kafka 推送数百万条消息，使用模式注册表 (SR) (Confluent)，而不是一直发送完整的模式。

架构在这张图片中是相关的：

事实是，我们希望避免在生产者想要发送消息或消费者将收到消息时一直调用 SR。

所以我这里要讲的是AvroSerializer和AvroDeserializer管理的“本地”缓存。我认为已经实现这一点的贡献者应该很好地完成了这项工作。

但是由于牵连的调用数量，我只想确定我在上述架构中用红色写的内容

我没有找到任何功能来管理这个“本地”缓存，你知道检查它的内容、大小......

非常感谢！

我正在通过 Confluent 平台运行 Kafka。我已经按照提到的步骤进行操作，java.net.BindException: Address already in use

根据此处的文档，https://docs.confluent.io/2.0.0/quickstart.html#quickstart

启动动物园管理员，

启动卡夫卡，

接下来当我运行schema-registry命令时，

我观察到错误，java.net.BindException: Address already in use

我在 macbook 中本地运行所有这些。somoene 可以帮我解决这个address already in use错误吗？

控制台日志：

请帮助解决此错误。

谢谢，

当我运行命令时，ps aux | grep schema-registry

我正在使用 Confluent 3.3.0。我的意图是使用 kafka-connect 将来自 Kafka 主题的值插入到 Oracle 表中。我的连接器适用于我使用 avro 控制台生成器生成的 avro 记录，如下所示：

我插入如下值：

我想要实现的是使用 Java 应用程序，使用对象插入相同的数据。下面是我的生产者代码：

以下是飞行画外音：

最后，序列化器：

但我的理解是，需要定义类似模式的东西，并使用一些 avro 序列化程序来获取确切的数据，就像我使用avro console consumer所做的那样。我已经浏览了一些示例代码，但没有一个对我有用。

编辑

我尝试了以下代码。但是在 avro 控制台消费者中什么也没有出现。

当我使用 kafka 流加入 Windows 时，自动创建一个 avro 模式

像这样 “* KSTREAM-JOINTHIS-0000000125-store-changelog-value”**

我想知道，为什么这可以创建 avro 模式？

有我的代码：