问题标签 [kafka-rest]

我正在尝试将 AWS MSK 与 Confluent 提供的 RestProxy 集成。我创建了 AWS MSK 集群，没有进行身份验证，并在同一个 VPC 上部署了 confluent 提供的 RestProxy。在休息代理上为引导服务器提供明文 Zookeeper 连接和 SSL 后，当我点击时

我从休息代理那里得到 404。如果我打

资源以空 json {} 响应

我不确定我错过了什么。

我正在使用 confluent-5.5.1。我正在按照https://docs.confluent.io/platform/current/kafka-rest/api.html#consumers链接上的说明进行操作，但出现“未找到消费者实例”错误。

要求：

回复：

我的休息代理服务日志：

Endpoint : http://localhost:8082/v3/clusters/cluster-id/topics

post body : { "topic_name": "topic-X", "partitions_count": 1, "replication_factor": 1, "configs": [ { "name": "cleanup.policy", "value": "compact" }, { "name": "compression.type", "value": "gzip" } ] }

Exception : java.lang.NullPointerException: Cannot invoke "io.confluent.kafkarest.entities.v3.CreateTopicRequest.getTopicName()" because "request" is null at io.confluent.kafkarest.resources.v3.TopicsResource.createTopic(TopicsResource.java:137)

I'm trying to wrap the Confluent kafka proxy api in one class that will handle producing and consuming.
Following this link: https://docs.confluent.io/platform/current/kafka-rest/api.html I tried to implement it as follows:

but I ended up working with 2 versions of the api (v2/v3) cause I didn't find some api's in one implementation and vise versa...
For example I didn't find how to create topic in v2, so I implemented it with v3.

My issue now is with the send method, I'm getting Internal server error and I can't find why!
Maybe because the create topic was done with v3 and I'm trying to produce messages with v2.

我正在尝试对 AWS MSK 集群使用 kafka 休息代理。

MSK 加密详细信息：

集群内

TLS 加密：启用

客户和经纪人之间

TLS 加密：启用

明文：未启用

我在 MSK 上创建了主题“TestTopic”，然后在与 MSK 相同的 VPC 中创建了另一个 EC2 实例，以作为 Rest 代理。以下是来自 kafka-rest.properties 的详细信息：

我还创建了包含以下内容的 rest-jaas.properties 文件：

然后使用以下方法设置 java.security.auth.login.config：

在此之后，我使用以下方法启动了 Kafka 休息代理：

但是，当我尝试通过从邮递员调用服务来在 TestTopic 上放置一个事件时： POST: http://IP_of_ec2instance:8082/topics/TestTopic 我收到 500 错误。但在 EC2 实例中，我可以看到错误：

我还可以在 ProducerConfig 值中看到 sasl.jaas.config = null 的值。

有人可以帮我解决这个问题。提前致谢！

Kafka Connect 源和接收器连接器提供了几乎理想的功能集，无需编写任何代码即可配置数据管道。就我而言，我想用它来集成来自公共 Internet 上的多个数据库服务器（生产者）的数据。

然而，一些生产者无法直接访问 Kafka 代理，因为他们的网络/防火墙配置只允许到特定主机（端口 443）的流量。不幸的是，我无法真正更改这些设置。

我的想法是使用 Confluent REST 代理，但我了解到 Kafka Connect 使用 KafkaProducer API，因此它需要直接访问代理。

我发现了几个可能的解决方法，但没有一个是完美的：

1. SSH 隧道 - 如：通过 SSH 隧道从 Kafka 集群消费
2. 使用 REST 代理，但用自定义生产者替换 Kafka Connect，在我们如何在防火墙/代理后面配置 kafka 生产者？
3. 使用 SSHL 解复用器将流量路由到代理（但只有一个代理）

有没有人遇到过类似的挑战？你是怎么解决的？

需要您的帮助来了解如何使 kafka rest 调用可空十进制字段。该字段的架构定义如下：

{"name":"column","type":["null","{"type":"bytes","logicalType":"decimal","precision":4,"scale":0}]} "]}

对于原始类型，它很容易发送值，因为我们必须随调用一起发送数据类型。因此，对于可为空的字符串字段，空值类似于 "column": {"null" : null} 而不是空值类似于 "column": {"string": "ABC"}。

但在这种情况下，我需要发送逻辑类型列的值，并且通过以格式发送不起作用 - “column”：{“bytes”：“100.2”}

对此的任何直接帮助都将受到高度赞赏。

谢谢！

我正在尝试使用 kafka-rest 接口将 Protobuf 模式发布到模式注册表：

我收到此错误：

问题：指示此工作的媒体类型的正确方法是什么？

我需要在 kubernetes 上启动 Kafka Rest Proxy。当我使用 docker-compose 在本地机器上使用本地 Kafka 代理和 zookeeper 启动它时，它运行良好，我得到了最后一个日志输出，表明 Kafka Rest 正在运行并等待请求。当我在 kubernetes 上启动它时，我只得到以下输出：

然后 pod 以错误代码 1 退出并处于 CrashLoopBackOff 状态。

对于我设置的环境变量KAFKA_REST_BOOTSTRAP_SERVERS，格式为PLAINTEXT://{host1},PLAINTEXT://{host2}，KAFKA_REST_LOG4J_ROOT_LOGLEVEL = DEBUG但它没有显示任何额外的输出。

有没有人有同样的问题？

我正在尝试在 Kafka 休息代理上应用 mTLS 安全性，但没有运气。我正在寻找的模型如下。

浏览器--https://host:443/--> Kafka Rest代理--kerberos--> Kafka Brokers

卡夫卡经纪人的休息代理工作正常，但休息代理的客户端仅适用于 http://host:port/

我的 Kafka-rest.properties 如下。

有人可以指出我在这里缺少什么吗？