问题标签 [kafka-rest]

我正在尝试在 kafka 主题上发布一些数据，其中数据的 avro 模式包含联合类型。我在 android 上使用 kafka-rest 代理来生成消息。

鉴于数据包含联合类型；pojo 到 json 转换器似乎都不起作用。

我在 Schema 注册表中注册了一个 Schema，如下所示：

以上所有字段都是可选的

当我通过 Rest Proxy 发送数据负载时，如下所示：

{ "":val1, Prop2:val2, Prop3:val3 }

我希望架构注册表会引发错误。但这并没有发生。数据通过接收器连接器，第一个键/值对被忽略。

这是预期的行为吗？如果属性不存在，我如何强制出错（该属性是可选的，但是当我们发送空白属性时，它应该会抛出错误）。

我正在尝试使用一些自定义标头向 Kafka 发送消息，但我找不到这样做的方法。根据文档，POST /topics/(string: topic_name)API 没有将自定义标头附加到消息的方法。

有谁知道通过融合的 Kafka REST API 发送带有自定义标头的 Kafka 消息的方法？

这是我发送到 REST 代理的示例请求正文

我正在尝试使用 Confluent Kafka REST 代理从我的一个主题中检索 Avro 格式的数据，但不幸的是我收到了反序列化错误。我正在使用以下命令查询 Kafka REST 代理

我得到了回应

Kafka Rest Proxy 服务器上的日志是：

数据是使用 KafkaAvroSerializer 生成的，并且模式存在于模式注册表中。另请注意，通过在 CLI 上使用 avro-console-consumer 可以读取数据。

有谁知道如何解决这个问题？

我会将 kafka-rest-proxy 融合解决方案与 apache kafka 2.0.0 集成

有人可以解释我如何为我的集群只安装 3 个节点 kafka 和 3 个 znodes 的 kafka rest 代理吗？

我正在使用 confluent REST API 代理调用 Kafka。我正在读取一个 CSV 文件，从那里的所有记录（大约 400 万条记录）中创建一个对象，并向 REST 代理发送请求。我不断收到OutOfMemory异常。

确切的异常消息是：

Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "kafka-producer-network-thread | producer-81"

我只有一个 REST 代理服务器实例，作为 docker 容器托管。环境变量设置为：

JAVA_OPTIONS=-Xmx1g

其他配置：

CPU - 1 Memory - 1024

它在崩溃之前处理了大约 1,00,000 个。我尝试将其扩展到 4 个实例，同时将 CPU 增加到 3，内存也增加到 2046 mb。然后它处理大约 5,00,000 条记录。

阅读 csv 后，我以 5k 条记录为单位调用 Kafka 端点。那是用Node写的。这是节点代码

我觉得我缺少一些配置，这些配置应该有助于解决这个问题，而不会增加实例数 > 1。

任何帮助将不胜感激。

我正在设计一个通知引擎（基于 Java），它将用作各种其他系统的一站式商店，以将通知发送到不同的渠道（推送通知、SMS、电子邮件等）。我希望构建这个系统以使用 Kafka 作为消息代理组件（还有许多其他组件）。

我的查询特定于应用程序推送通知（我们当前的系统使用来自发布的各个系统的基于 FCM 主题的订阅通知）。新的通知引擎将作为 Kafka 生产者工作，并定期向主题发布消息（如本系统的调度程序组件中配置的那样）。应用程序用户如何在通知发布后立即获得通知？我假设这里的应用程序必须作为 Kafka Consumer 工作。实现这一目标的最佳方法是什么？Kafka REST 代理是否可以让移动应用程序像 Kafka 消费者一样行动？移动应用程序承担所有消费者责任会不会超负荷？

我正在尝试将此消息发布到 Kafka Rest：

我在 rest-proxy.log 中有以下错误：

[2019-04-11 14:25:58,991] 错误未处理的异常导致内部服务器错误响应 (io.confluent.rest.exceptions.GenericExceptionMapper:38) org.apache.kafka.common.errors.SerializationException：注册 Avro 架构时出错：“long” 原因：io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException：正在注册的模式与早期模式不兼容；错误代码：409

为什么会这样？？？我不明白为什么

是否正在触发新模式的创建？...

顺便提一句

工作正常。

我正在使用 Kafka 1.1 和 Kafka Rest Proxy 4.1.2。我一直在使用字符串键从 Kafka 流中键入记录。我想使用 Rest Proxy 插入要加入的记录，但是键被转义的引号放在它们周围。

我正在发送一个导致问题的POST请求。/topics/{someTopic}Content-Type: application/vnd.kafka.json.v2+json

和key_schemaContent-Type: application/vnd.kafka.avro.v2+json类型：字符串，键周围没有多余的引号，但我宁愿发送 json 值。

这就是我发送到/topics端点的内容。

当我在 Kafka 流中流式传输数据时，密钥以 形式出现\"abc\"，并且显然没有与带有字符串键的记录连接abc。

有没有办法用 json 值指定一个键模式，这样我的键就不会在它们周围得到转义引号？

我正在尝试将数据（有效负载）从 BitBucket 发布到我的 Kafka 实例，以防发生任何 BitBucket 事件（Pushed例如PR评论Forked等PR creating）AdditionModificationDeletion

所以，我已经设置了 Kafka rest-proxy，但我无法将数据从 BitBucket 发送到 Kafka，因为使用Kafka REST 代理Content-Type发布数据所需的是，并且 BitBucket 的有效负载的标头设置为（我使用的是默认的 BitBucket webhook在级别配置）。application/vnd.kafka.json.v2+jsonapplication/jsonRepository

我在存储库级别配置了 Webhook 是这样的：

这是从 BitBucket 的 Webhook 触发的请求：

我从 Kafka 的 REST 代理得到的响应是这样的：

我考虑过以下解决方案：

1. 我可以编写一个简单的 Web 服务器，充当两个系统（BitBucket 和 Kafka）之间的中介。服务器可以充当中间代理，从 BitBucket 的 webhook 获取传入请求，并Kafka REST proxy使用适当的标头 ( application/vnd.kafka.json.v2+json) 将其发布到。

   该解决方案有效，但缺点是维护另一台服务器（由我编写）。此外，由于 BitBucket 开始频繁触发 webhook（高峰时段每秒 100 次，因为我们只有一个 BitBucket 实例并且我们有大约 3500 名开发人员） ，因此它可能会遇到可伸缩性问题。

2. 我可以使用一个 BitBucket 插件，它可以作为 awebhook并且允许我动态配置和更改标头。这是我正在努力实现的解决方案。也不需要维护额外的服务器。

是否有一个 BitBucket 插件，我可以从市场上安装它，它可以让我在触发 webhook 时配置标头？

PS - 很抱歉写了这么长的问题。这两个解决方案只是我的想法。如果有更好的解决方案将 BitBucket 与 Kafka 集成，我不会感到惊讶。另外，如果问题不清楚，请告诉我。在这种情况下，我会尝试以不同的方式澄清我的问题。

谢谢