问题标签 [message-hub]

我正在使用 kafka python 客户端将消息推送到 Message Hub，但注意到在运行我的应用程序一段时间后它会停止向 Message Hub 发送消息。

然后我在我的日志文件中注意到以下内容：

我更新了我的代码以添加retries=5：

我的代码从 python 烧瓶应用程序运行。每个传入的特定类型的请求都会调用该send_message()方法。

以下是来自 Bluemix 的相关日志行。我可能错过了一两行复制和粘贴，但希望这足以弄清楚发生了什么：

我的话题存在。我的完整客户端代码在这里：https ://github.com/snowch/movie-recommender-demo/blob/effc981cc9f799c41952719619f693172eebcd6a/web_app/app/messagehub_client.py

任何最受赞赏的指针...

这是我正在运行的代码片段：

基本上，这可以在一台机器上成功运行，但不能在另一台机器上运行。我在不工作的机器上收到的错误如下：

不确定此问题是否与此问题有关：为什么 kafka-python 无法连接到 Bluemix 消息中心服务？...不要这么想，因为确切的代码在一台机器上运行，而另一台机器上却没有。任何帮助表示赞赏！

我尝试使用 Liberty for Java 应用程序在 Bluemix 上建立与 kafka-message-hub 的连接。我遵循了文档和 git 存储库示例。它可以正常创建/使用jaas.conf文件并表示它已登录。但是，当写入消息时，它会显示“在 60000 毫秒后更新元数据失败”。为了写消息，我必须在server.xml文件中添加以下代码：

但是，文档说（在“如何将 Kafka 客户端从 0.9.x 迁移到 0.10.x”下）：

1. 删除 Message Hub 登录 Jar 模块。
2. 将您的jaas.conf文件更改为以下代码：

KafkaClient { org.apache.kafka.common.security.plain.PlainLoginModule required serviceName="kafka" username="<your username>" password="<your password>"; };

3. 将以下行添加到您的消费者和生产者属性中：
   sasl.mechanism=PLAIN

我在这里错过了什么吗？似乎说明说我不再需要这个模块了，但是没有它我就无法让它工作。

我正在查看蓝色混合消息集线器的 node-js express 示例，并且对为什么需要从我的服务器轮询消息集线器感到困惑。我认为 pub-sub 模型背后的整个想法是我不必（过度）加载我的服务器来轮询消息服务来确定是否存在新消息供我使用。在提供的示例中，app.js 文件中的第 211-213 行包含以下内容：

现在，我的服务器每 250 毫秒轮询一次消息集线器，而我想要完全避免轮询模型，并在存在消息供我使用时由消息集线器通知。

这个问题与为什么我需要轮询消息集线器有关，与该答案链接的文档显示Kafka将支持“长轮询”概念，但是现有的nodejs支持消息集线器没有明确的方法来实现这样的机制。为 nodejs 提供的演示应用程序仅使用 250 毫秒的计时器间隔来处理从服务器检索消息。我想用更复杂的长轮询方法替换它，使用 Kafka 支持长轮询方法： To avoid this we have parameters in our pull request that allow the consumer request to block in a "long poll" waiting until data arrives 但是现有实现似乎不允许配置任何此类参数，也不清楚必要的参数是什么是。get函数的原型定义为：

所以没有配置选项。或者，您能否提供一个链接，该文档定义了在 message-hub.js 模块中实现的那些 URL 的 URL 和可用选项？

我看到了一个问题，即我的主题中的所有消息都被我的消费者重新阅读。我只有 1 个消费者，我在开发/测试时打开/关闭它。我注意到有时在几天不运行消费者之后，当我突然再次打开它时，它会重新读取我的所有消息。

clientid 和 groupid 始终保持不变。我明确调用 commitSync，因为我的 enable.auto.commit=false。我确实设置了 auto.offset.reset=earliest，但据我了解，只有在服务器上删除偏移量时才应该启动。我正在使用 IBM Bluemix 的 MessageHub 服务，所以可能会自动删除偏移量？

有没有人有任何线索/想法？

谢谢。

我已经阅读了其他线程，并且通过使用新的组 ID 解决了这个问题，但是我想了解可能导致这种情况的原因。

我有一个包含 16 个分区的主题，我设置了 session.timeout.ms=30000 和 max.poll.interval.ms=30000000。

我运行我的程序，然后按 ctrl+c，所以它没有正确关闭。在我猜测 16 次之后，我陷入了这个重新加入的问题。session.timeout.ms 是心跳超时，所以 30 秒后它应该踢我的消费者，我的分区应该“释放”对吗？还是只听我的 max.poll.interval.ms？

编辑：我仍然间歇性地收到这个错误，当它发生时我必须重新启动我的所有消费者。即使我的消费者运行良好然后他们都开始陷入重新加入（没有添加/删除消费者），也会发生这种情况。这是我在新消费者卡在该状态后尝试连接它时的错误日志：

https://pastebin.com/AXJeSHkp

这些是我认为相关的第一条和最后一条消息。以下是我设置的相关超时：

我也应该设置 heartbeat.interval.ms 吗？这是消费者在某个后台线程中自动将心跳发送到代理的间隔（我已经阅读了文档，但由于某种原因我无法完全理解它）？

我有一个连接 Bluemix Message Hub 的自由应用程序。它在本地运行良好，但是当我从 Bluemix Liberty 运行 liberty 应用程序时它永远无法运行。我正在关注https://github.com/ibm-messaging/message-hub-samples/tree/master/kafka-java-liberty-sample中的示例，您在运行时更改 server.xml 中的用户名/密码。

经过反复试验，我意识到当我在 server.xml 中硬编码我的 kafka 用户名/密码时，BluemixLiberty 可以正常连接到 MsgHub。当我像示例中那样“热交换”kafka 用户名/密码时，它不起作用并且永远不会连接到 kafka（即使我重新启动服务它也不会重新连接）。我已经通过使用 CF 登录到我的服务进行了检查，并且 sever.xml 确实正确交换了用户名/密码。

Bluemix Liberty 是否支持此功能？如果不是，我应该如何正确使用 vcap-services，这样我就不会对我的生产用户/密码进行硬编码？

我看到我当前实例未订阅的主题的 kafka 警告。我们使用相同的 kafka 代理，每个开发人员都有自己的主题。但是，我们都使用在属性文件中指定的相同默认 groupID/clientid。

这是预期的吗？即使每个人都有自己的主题，我是否需要单独的组/客户 ID？

其次，我不太确定这个错误意味着什么：心跳协调器线程无法获取 2 个主题的元数据，但这些主题存在，只是这些主题的消费者没有运行。

我一直在尝试查找通过 C# Apache Kafka Confluent API 连接到 IBM Message Hub 的文档，但没有成功。github 上的官方 repo 没有 C# 的示例。有没有人能够使用 C# 与 ibm 消息中心通信，如果是的话，你可以分享这个过程。

谢谢。

更新：我已成功与 IBM Message Hub 进行通信。

图书馆：

1. librdkafka -... 0.11.0-RC2

2. 证书来自：https ://curl.haxx.se/docs/caextract.html

3. Confluent.kafka.dll Confluent.Kafka 0.11.0-RC1

配置：

.net 版本：4.5.2

希望它可以为某人节省时间。

感谢 Edoardo Comar 指导我获取急需的信息。