问题标签 [kafka-rest]

是否可以修改/自定义响应内容？假设我们已经设置了 kafka rest 代理来接收来自外部服务的数据，并且我们不想泄露敏感信息，比如存储在哪个分区和偏移数据中。我们只想回复说记录已处理/已接收（或未处理）。

当前成功响应（200）

可以将上述响应定制或截获为以下响应吗？

我们正在使用 Confluent Rest Proxy 与 Kafka 通信，需要测试各种数据。我们正在使用 Rest Proxy 来允许供应商与我们的 Kafka 系统进行通信。

Avro 模式中的一个字段具有十进制的逻辑类型。为了简单起见，让我们假设这里显示的模式：

通过 Java 生产者编写主题很容易，使用 Avro 工具生成适当的类文件。但是当尝试使用 Rest Proxy 时，我们必须传递如下值：

这是从通过 Java 生产者创建的记录中复制的，然后从主题中下载。但在amount现场，我们希望能够传递一个值，例如123.45. 我们主要使用 Postman 来发送数据。有没有办法使用逻辑十进制字段来做到这一点，而不必先创建和序列化数据以查看表示，例如\u0001ã？

kafka rest proxy指定要获取的记录数。

GET /consumers/(string:group_name)/instances/(string:instance)/records

获取使用订阅/分配 API 之一指定的主题或分区的数据。

此 api 用于从主题中获取记录。是否有一个方便的参数来指定我想要获取的记录数。

例如，我想从某个偏移量中获取 10 条记录。
这个 api 只有两个查询参数，timeout和max_bytes。没有像limit_num这样的参数。

POST /consumers/(string:group_name)/instances/(string:instance)/positions

覆盖消费者将用于下一组要获取的记录的获取偏移量。

我可以使用这个 api 来设置我想要获取的起始偏移量。

如何为 kafkarestproxy 创建信任库文件接下来如何为 rest 代理请求 mtls 身份验证。如何测试 mtls 身份验证？

当我向https://192.168.1.98:8083/发出请求时，我收到“err_connection_closed”错误。我可能做错了什么？

我的卡夫卡-rest.properties：

我有一个应用程序的两端：

1. Python Flask 后端，以正常方式使用 Kafka 进行通信。
2. 用 Go 编写的机器代理，安装在客户端环境中，仅通过 Kafka REST 代理与 Kafka 通信。

现在的问题是这两个端可以通信吗？例如，我的机器代理可以通过 REST 代理使用来自 Kafka 的消息，并以正常方式从另一端产生消息吗？还是两端都需要使用Kafka REST Proxy？

我的配置：

错误日志：

我在这里阅读了如何设置 Kafka REST API：https ://docs.confluent.io/platform/current/installation/docker/config-reference.html#crest-long-configuration

还找到了这个配置示例：https ://github.com/conduktor/kafka-stack-docker-compose/blob/master/full-stack.yml

似乎问题是：http ://www.slf4j.org/codes.html#multiple_bindings 如果我启用模式注册表 - 它在 thr 日志中有同样的问题。

我如何仅在 docker-compose.yml 文件中使用环境变量来解决上述问题？此处描述的内容：https ://docs.confluent.io/platform/current/installation/docker/operations/logging.html

更改了我的配置以显示导出变量：

完整的日志说没有设置listeres、zookeeper.connect和bootstrap.servers：

为什么 docker-compose.yml 环境变量不适用于配置？

我能够以某种方式运行 6.0.0 版本：

我在这里找到了具有预定义模式 id 的创建消息格式： https ://docs.confluent.io/platform/current/kafka-rest/api.html#post--clusters-cluster_id-topics-topic_name-records 它略有不同从 v2. 我没明白 key.data 是什么意思？

教程仅显示版本 v2 https://docs.confluent.io/platform/current/tutorials/examples/clients/docs/rest-proxy.html#basic-producer-and-consumer

我透露教程中的教程模式版本参考在kafka version3 中不起作用。

另外，如果我尝试通过 kafka-rest v2 从 cp-kafka-rest:6.2.1 而不是 cp-kafka-rest:6.1.0 发送相同的教程，我会收到相同的结果

我得出的一个假设是模式注册表不依赖于 kafka-rest 版本——它正确吗？

教程：

https://docs.confluent.io/platform/current/tutorials/examples/clients/docs/rest-proxy.html#basic-producer-and-consumer https://docs.confluent.io/platform/current/kafka -rest/api.html#post--clusters-cluster_id-topics-topic_name-records

消息：

答案：

我在用 Go 开发的应用程序中使用 Rest 代理，使用主题的步骤是：

1. 创建消费者实例
2. 订阅主题
3. 获取记录
4. 并删除消费者实例

现在我的问题是我想让消费者始终在 goroutine 轮询主题中运行，我是否需要始终遵循每个步骤，或者我可以先执行第 1 步和第 2 步，然后继续在第 3 步循环，而不是删除消费者实例？

我正在尝试将 kafka 消息发送到 android 应用程序。

我在游戏后期才意识到，唯一的方法是使用 Kafka Rest Proxy。

我已经设置好了，但是我的密钥没有被引用，导致反序列化错误。

和/bin/kafka-console-consumer --bootstrap-server localhost:9092 --topic my_topic --property print.key=true --from-beginning

我收到 truck/6084/position {"timestamp":1636987919544,"truckId":6084,"driverId":36,"routeId":1177752383,"eventType":"Normal","correlationId":"597220341814695046","latitude":24.7700961485576,"longitude":46.6207663588065}

我已经将这些消息通过典型的消费者几乎一行一行地从融合中运行到 PC 应用程序，没有任何问题。

我的确切错误是

尝试使用“jsonschema”时

任何见解将不胜感激。

克里斯

编辑：