问题标签 [confluent-schema-registry]

所以有一个 Kafka 集群，我们在它上面有一个 Schema 注册表来验证主题的模式。出于某种维护原因，如果模式注册表关闭，Kafka 将在这段时间内停机，并且它不会接受任何新的传入数据请求？

我们有一个处理消息生产和消费的服务器。我们有 4 台笔记本电脑，所有带有 confluent 的 Mac 都运行相同的命令行......

./kafka-avro-console-consumer --from-beginning --bootstrap-server 0.0.0.0:9092,0.0.0.0:9092 --topic topicName --property schema.registry.url= http://0.0.0.0 :8081

4 台笔记本电脑中有 3 台在使用消息时没有问题，但第四台不会。我已经确认我的笔记本电脑对 IP 地址具有相同的访问权限，我可以通过转到地址 /subjects 来查看架构注册表中的主题，并且我在控制台和日志中都没有收到任何错误。

我完全不知道还要检查什么。有没有其他人遇到过这种情况并可能找到解决方案？

问题：在使用带有 Avro 序列化器和反序列化器的 Kafka Connect JDBC 源设置 Kafka 管道后，当我尝试使用 Kafka Streams Java 应用程序将该数据读入 KStream 时，我会收到以下错误。

org.apache.kafka.common.errors.SerializationException: LongDeserializer 接收到的数据大小不是 8

我试图尽可能地遵循现有的例子，但有些事情是没有意义的。我将在下面包含所有代码/附加信息，但这里有几个问题......

1. 我目前理解的最大差距之一是 Avro 记录的“KEY”使用的是什么？我（在运行时）出错的那行与我告诉 KStream 密钥是 LONG 的事实有关，但是当检索 Avro 记录时，长度小于 8（预期长度为 LONG类型）。
   当我设置我的 JDBC 源时，那里没有任何东西可以识别密钥是什么——而且我在文档中没有看到任何东西可以让我相信我可以指定密钥，尽管我已经尝试过：

   /li>

有了以上内容，我得到了运行报告的以下模式：

这是它的输出：

要更漂亮地查看该架构：

再说一次，我没有看到任何表明上面的任何特定字段将成为记录的关键的任何东西。

所以然后我进入 Kafka Streams，我尝试将这些数据带入 KStream ......它会爆炸......

所以，事情就是这样，因为我知道存储在幕后的数据是 SQL Server 中的 BIGINT 并且映射到 Java 中的 LONG，所以我将 KStream 的键类型设为 Long，然后我使用 Serdes.Long() 反序列化器KStream 构建器的参数。

调试时，我看到原始记录的长度为 7，这就是它抛出错误的原因。显然 Avro 以更好地压缩的方式序列化事物？我不知道。无论如何，问题是我什至不知道它认为它实际上使用的是什么键？！那么谁知道 - 也许我对 Long 的假设是不正确的，因为它实际上并没有使用 ApplicationID 作为密钥？为什么我什至会认为它是？！

对此的任何帮助将不胜感激。我知道那里有很多信息，但简而言之..

1. 使用 JDBC Kafka 连接将数据推送到主题
2. 数据正在成为主题 - 我可以通过控制台看到它
3. 试图将这些数据推送到流中，以便我可以对数据做一些很棒的事情，但由于 Serdes 与 Avro Record 不兼容，因此尝试填充流时它会爆炸

更新 1：根据下面 Randall 的建议，我尝试了 SMT（单消息转换），现在我每条记录都有一个密钥，这是朝着正确方向迈出的极好一步，但由于某种原因它没有出现强制转换为 Long (INT64) 有任何实际效果。我已经使用 SMT 截取了连接器配置的一些屏幕截图，生成的记录（现在有一个密钥！）以及我在 Kafka 流中看到的相同错误：

致力于设置从我们的 RDS Postgres 9.6 运行到 Redhift 的 Kafka。使用https://blog.insightdatascience.com/from-postgresql-to-redshift-with-kafka-connect-111c44954a6a上的指南，我们已经设置了所有基础设施，并且正在努力全面设置 Confluent。我收到 ava.lang.IllegalArgumentException 的错误：组数必须为正。尝试设置东西时。这是我的配置文件：

完整错误：

/usr/local/confluent$ /usr/local/confluent/bin/connect-standalone /usr/local/confluent/etc/schema-registry/connect-avro-standalone.properties /usr/local/confluent/etc/kafka- connect-jdbc/source-postgres.properties SLF4J：类路径包含多个 SLF4J 绑定。SLF4J：在 [jar:file:/usr/local/confluent/share/java/kafka-serde-tools/slf4j-log4j12-1.7.6.jar!/org/slf4j/impl/StaticLoggerBinder.class] 中找到绑定 SLF4J：在 [jar:file:/usr/local/confluent/share/java/kafka-connect-elasticsearch/slf4j-simple-1.7.5.jar!/org/slf4j/impl/StaticLoggerBinder.class] SLF4J 中找到绑定：找到绑定在 [jar:file:/usr/local/confluent/share/java/kafka-connect-hdfs/slf4j-log4j12-1.7.5.jar!/org/slf4j/impl/StaticLoggerBinder.class] SLF4J：在 [ jar：文件：/usr/local/confluent/share/java/kafka/slf4j-log4j12-1.7.21.jar！http://www.slf4j.org/codes.html#multiple_bindings解释一下。SLF4J：实际绑定的类型为 [org.slf4j.impl.Log4jLoggerFactory] ​​[2018-01-29 16:49:49,820] INFO StandaloneConfig 值：access.control.allow.methods = access.control.allow.origin = bootstrap。服务器 = [localhost:9092] internal.key.converter = 类 org.apache.kafka.connect.json.JsonConverter internal.value.converter = 类 org.apache.kafka.connect.json.JsonConverter key.converter = 类 io。 confluent.connect.avro.AvroConverter offset.flush.interval.ms = 60000 offset.flush.timeout.ms = 5000 offset.storage.file.filename = /tmp/connect.offsets rest.advertised.host.name = null 休息。广告.port = null rest.host.name = null rest.port = 8083 task.shutdown.graceful.timeout。305] INFO Herder 已启动 (org.apache.kafka.connect.runtime.standalone.StandaloneHerder:72) [2018-01-29 16:49:50,305] INFO 正在启动 REST 服务器 (org.apache.kafka.connect.runtime.rest .RestServer:98) [2018-01-29 16:49:50,434] INFO jetty-9.2.15.v20160210 (org.eclipse.jetty.server.Server:327) Jan 29, 2018 4:49:51 PM org. glassfish.jersey.internal.Errors logErrors 警告：已检测到以下警告：警告：org.apache.kafka.connect.runtime.rest.resources.ConnectorsResource 中的（子）资源方法 listConnectors 包含空路径注释。警告：org.apache.kafka.connect.runtime.rest.resources.ConnectorsResource 中的（子）资源方法 createConnector 包含空路径注释。警告：org.apache.kafka.connect.runtime.rest.resources 中的（子）资源方法 listConnectorPlugins。ConnectorPluginsResource 包含空路径注释。警告：org.apache.kafka.connect.runtime.rest.resources.RootResource 中的（子）资源方法 serverInfo 包含空路径注释。[2018-01-29 16:49:51,385] 信息开始 oejsServletContextHandler@5aabbb29{/,null,AVAILABLE} (org.eclipse.jetty.server.handler.ContextHandler:744) [2018-01-29 16:49:51,409 ] INFO Started ServerConnector@54dab9ac{HTTP/1.1}{0.0.0.0:8083} (org.eclipse.jetty.server.ServerConnector:266) [2018-01-29 16:49:51,409] INFO Started @2019ms (org. eclipse.jetty.server.Server:379) [2018-01-29 16:49:51,410] INFO REST 服务器正在监听http://127.0.0.1:8083/，广告网址 http://127.0.0.1:8083/ connect.jdbc.JdbcSourceConnector key.converter = null name = source-postgres tasks.max = 16 value.converter = null (org.apache.kafka.connect.runtime.SourceConnectorConfig:180) [2018-01-29 16:49: 52,209] 错误在连接器错误后停止 (org.apache.kafka.connect.cli.ConnectStandalone:102) java.lang.IllegalArgumentException：组数必须为正。在 org.apache.kafka.connect.util.ConnectorUtils.groupPartitions(ConnectorUtils.java:45) 在 io.confluent.connect.jdbc.JdbcSourceConnector.taskConfigs(JdbcSourceConnector.java:123) 在 org.apache.kafka.connect.runtime .Worker.connectorTaskConfigs(Worker.java:193) 在 org.apache.kafka.connect.runtime.standalone.StandaloneHerder.recomputeTaskConfigs(StandaloneHerder.java:251) 在 org.apache.kafka.connect。

我们在 RDS Postgres (9.6) 和 Redshift 之间使用 DMS。它一直在失败，而且很悲惨，而且几乎在这一点上几乎是非常昂贵的，所以我们正在将其作为一种可能的解决方案。我在这里有点困难，真的很想在这方面得到一些帮助。

我在 EC2 上设置了从 kafka 读取并写入 S3 的融合（4.0）连接器。

独立尝试进展顺利：

bin/connect-standalone 等/standalone/example-connect-worker.properties 等/standalone/example-connect-s3-sink.properties

然而，分布式版本不断失败

我只是想首先使用连接器类等于 FileStreamSinkConnector

接收器 conf 文件如下所示：

非常感谢！

我一直在写一个自定义的 kafka 连接器插件。为此，我必须从元数据文件动态创建模式，然后读取数据文件中的每条记录并将单个记录发布到主题。下面是创建模式的示例代码：

创建单个字段

创建记录

我有所有字段的逻辑类型，我也将这些字段写入 avsc 文件。在文件中，它显示为：

但是，当我在http://localhost:8081/subjects/myvalue/versions/1调用架构注册表端点时，它显示如下：

我已经调试了足够多，但无法取得任何进展。我在这里做错了什么？

我知道我会带着任何消息来找你，但我一直在解决一个可能是我的错的问题，实际上我不知道解决方案是什么。我正在使用 Confluent 平台（4.0.0 开源版本）的独立安装，以演示如何将平台用于特定用例。试图展示使用模式注册表的价值，我面临以下问题，使用 Postman 发布新模式。

请求是：

响应是： {"error_code":42201,"message":"Input schema is an invalid Avro schema"}

查看文档并在谷歌上搜索了很多之后，我没有选择。有什么建议吗？谢谢你的时间R。

我在 Spark Structured Streaming 中使用 Kafka Source 来接收 Confluent 编码的 Avro 记录。我打算使用 Confluent Schema Registry，但与 spark 结构化流的集成似乎是不可能的。

我已经看到了这个问题，但无法让它与 Confluent Schema Registry 一起使用。使用 Spark 2.0.2（结构化流）从 Kafka 读取 Avro 消息

当我启动 Confluent 时，Schema-registry 失败，导致进程无法成功完成。这是我得到的回应：

当我尝试单独运行这些进程时，zookeeper 运行没有问题。但是，当我启动 kafka 时，zookeeper 显示以下错误：

然后，当我尝试运行 Schema 注册表时，我遇到了大量的错误列表。我确信这些错误都指向一件小事。以下是一些错误（许多错误在同一条长消息中重复）：

1.

2.

3.

4.

一些错误因位置而略有不同，但在大多数情况下，这 4 个错误会被打印数十次。

我尽力确保其他进程没有使用任何端口。我还停止并销毁了我之前创建的所有融合实例。我以前在这台计算机上玩过 Kafka，所以我推测这可能与它有关，但我确保关闭所有过去的 zookeeper 和 kafka 实例。

我尝试在另一台计算机上运行 confluent 并没有遇到任何问题。有谁知道可能是什么问题？我可以发送整个错误消息并提供任何其他详细信息。

提前致谢！

我在 MacOS 上运行本地版本的 Confluent (4.0)，在启动它并创建主题newtopic并转到http://localhost:18081/subjects/newtopic之后 - 出现以下错误：
{"error_code":405,"message":"HTTP 405 Method Not Allowed"}

去http://localhost:18081/工作正常。

我应该检查什么以及如何排除故障？

谢谢。