问题标签 [strimzi]

我在 Kubernetes 中有一个使用 Strimzi 创建的 Kafka 集群。

集群已创建并启动，我能够创建主题并从主题生产/消费。问题是，如果我执行到 Kafka 代理 pod 之一，我会看到间歇性错误

在检查了这些 IP [10.240.0.35, 10.240.0.159, 10.240.0.4,10.240.0.128] 后，我发现它们与 kube-system 命名空间中的 pod 相关，这些命名空间是作为 Kafka 集群部署的一部分隐式创建的。

知道有什么问题吗？

我已经使用 Strimzi 在 Kubernetes 上配置了一个 Kafka 集群。使用这篇文章 - https://itnext.io/kafka-on-kubernetes-the-strimzi-way-part-2-43192f1dd831 但是我公司的安全服务不想在 UAT 中部署这个具有自签名的解决方案证书。您能否告诉我，我可以将我们的信任证书从受信任的冲浪中心导入到这个 Kafka 解决方案中吗？

我在我的 yml 文件中添加此部分，请参阅https://strimzi.io/docs/operators/master/using.html#kafka-listener-certificates-str 配置：brokerCertChainAndKey：secretName：es-kafka-secret 证书：certificate.crt密钥：证书.key

但是当我测试时我有错误

添加证书后如何检查连接？

请你帮助我好吗？谢谢。

我在我的 k8s 集群中启动并运行了桥接服务，所以我创建了与大使的映射，所以我的桥接 URL 是https://dev.ts.in。所以想创建主题，使用桥接 URL 发送消息。URL 适用于 curl 命令，那么方法是什么？将 url 与引导服务器一起使用是行不通的。任何帮助都会受到重视。

也许我错过了什么，如果是这样，请原谅我的无知。

这里我们有：

1. 我们在 Kafka 集群中使用 TLS 身份验证侦听器（这可以更改，我们可以添加新类型的侦听器）。
2. 从 Java 代码连接到 Kafka 主题时，我使用为 Kafka 用户生成的 SSL 证书。

如果我决定避免使用 SSL 证书，原因有两个：

1. 我将仅从受信任的 OpenShift 集群 POD 连接到 Kafka 主题
2. 为避免生产者/消费者端更新重新生成年度用户的 SSL 证书（因为 Kafka 生成的用户证书有效期为 1 年）

由于上述两个原因，KafkaUser 的 SCRAM-SHA-512 身份验证类型会是更好的（也是唯一的？）选择吗？或者 SCRAM-SHA-512 也需要 SSL 证书？

我看到的另一种方法是不进行身份验证，但我不确定 ACL 如何用于此类用户？我如何将用户正在连接的服务器信息传递给服务器。是否可以同时使用 ACL 而不是通过 SSL 证书或密码 Kafka 用户进行身份验证？

[UPD]环境建立在 Strimzi（OpenShift 中的 Apache Kafka 集群）之上

我正在尝试指定数据格式和 HTTP 标头，以确保使用 Strimzi Kafka 将有效请求提交给 Kafka Bridge。

消息格式 使用 /topics 端点发送消息时，您在请求正文中的 records 参数中输入消息负载。

记录参数可以包含以下任何可选字段：

消息头

消息键

消息价值

**Python中的示例代码**

导入请求

headers = {'content-type': 'application/vnd.kafka.json.v2+json', }

数据='{“记录”：[{“key”：“key-1”，“value”：“value-1”}，{“key”：“key-2”，“value”：“value-2” " } ] }'

response = requests.post('http://localhost:54304/topics/my-topic', headers=headers, data=data)

因此，我需要修改示例 JSON API 有效负载，使其与请求正文中 data = '{....}' 的数据格式相匹配。

示例 JSON API：http ://api.open-notify.org/astros.json

输出

{“消息”：“成功”，“数字”：11，“人”：[{“名称”：“迈克霍普金斯”，“工艺”：“ISS”}，{“名称”：“维克多格洛弗”，“工艺”：“国际空间站”}，{“名称”：“香农沃克”，“工艺”：“国际空间站”}，{“名称”：“野口宗一”，“工艺”：“国际空间站”}，{“名称” : "Mark Vande Hei", "craft": "ISS" }, { "name": "Oleg Novitskiy", "craft": "ISS" }, { "name": "Pyotr Dubrov", "craft": " ISS" }, { "name": "Thomas Pesquet", "craft": "ISS" }, { "name": "Megan McArthur", "craft": "ISS" },{“名称”：“Shane Kimbrough”，“工艺”：“ISS”}，{“名称”：“Akihiko Hoshide”，“工艺”：“ISS”}]}

请问如何将 JSON 有效负载格式化为用于生成消息的数据结构？

我尝试按照此文档在 docker-desktop k8s 上运行 srimzi 。

我设置了所有的东西。

并获取节点端口：

虽然本文档使用 minikube，但我可以通过 访问 docker-desktop 节点localhost，因此我尝试访问localhost:31314，但无法生成主题消息。看来我可以成功连接到代理，但不能连接到主题。

我用kafkacat检查过，主题本身已成功创建。

我不完全了解消息如何发送到主题的机制，那么接下来我应该检查哪一点？

我正在尝试设置使用我的 GoDaddy 证书作为 Kafka 的侦听器。使用这篇文章https://strimzi.io/docs/operators/in-development/using.html#kafka-listener-certificates-str。

我正在发送证书检查 openssl s_client -connect MY_IP:9094 -servername MY_IP

返回我的正确证书。但是当我尝试执行命令时出现错误

来自 Kafka 的日志

我正在尝试在 Kafka Connect 中实现静态成员资格。我们的 Kafka Connect 集群使用 Strimzi Kafka 算子部署在 K8S 上。

我尝试为工作人员（在 KafkaConnect yaml 中）放置以下配置：

在 HttpSinkConnector 类中，我有这个：

这给了 org.apache.kafka.common.errors.FencedInstanceIdException 一些日志 - ...07:07:32,631 错误 [Consumer instanceId=somethingsomething... 因为所有任务都有一些东西作为他们的 group.instance.id 虽然它应该有获取 Thread.currentThread().getName()。

我还尝试了以下方法（没有工作人员配置）：

这什么也没做（没有错误，日志中没有 instanceId），这意味着我把这个配置值放在了错误的地方。

那么如何在 Kafka Connect 上实现静态成员资格？

嗨，伙计们，连接没有附加磁盘，似乎使用节点磁盘代替。你知道strimizi中是否有增加或添加磁盘到kafka-connect的选项。

我正在使用 Kafka Strimzi 运算符在 Kubernetes 上运行 Kafka。我通过使用以下配置我的消费者来使用增量粘性重新平衡策略：

每次我在我的消费者组中扩展消费者时，该组中的所有现有消费者都会生成以下异常

线程“主”org.apache.kafka.common.errors.RebalanceInProgressException 中的异常：由于消费者正在为自动分区分配进行重新平衡，因此无法完成偏移提交。您可以尝试通过调用 poll() 来完成重新平衡，然后重试该操作

关于导致此异常的原因和/或如何解决它的任何想法？

谢谢你。