问题标签 [message-hub]

我一直在使用此链接中的 Message Hub 示例代码：https ://github.com/ibm-messaging/message-hub-samples

特别是，我一直在尝试使用 Kafka Java 控制台示例来提高生产者的吞吐量。我注意到这段代码中的文档：

// Synchronously wait for a response from Message Hub / Kafka on every message produced. // For high throughput the future should be handled asynchronously. RecordMetadata recordMetadata = future.get(5000, TimeUnit.MILLISECONDS); producedMessages++;

我已经关闭了稍后在代码中发现的线程睡眠，这也有助于提高吞吐量，但我希望我能在这个块中异步实现未来得到一些帮助。提前致谢！

我正在使用 Bluemix 消息中心服务。我需要向特定用户提供对特定主题的只读访问权限。问题是如何在 Bluemix 消息中心定义 ACL？Apache kafka 确实提供了方法（http://kafka.apache.org/documentation.html#security_authz），但这涉及与 Zookeeper 的交互。而且我无法找到连接到 Bluemix 消息中心服务背后的底层 zookeeper 的详细信息。欣赏回应。

当我尝试将 Bluemix Message Hub 与 Python 一起使用时，我不断收到错误和失败。任何想法为什么给出以下内容？

错误是“无法初始化 SASL 身份验证：代理不支持 SASL 握手（机制 PLAIN 需要）”

但是，当运行我的生产者和消费者时，它们似乎可以工作。产生消息并消费消息。

我在输出中看到与错误消息混合的正确消息。

我已经安装了 librdkafka 并按照说明确认一切正常：

https://github.com/ibm-messaging/message-hub-samples/blob/master/docs/librdkafka.md

当我运行 ./config 时，您可以在此处看到我的输出：

谢谢，亚伦

我正在尝试使用 curl 命令在 Message-Hub 中创建一个主题。
我跟着这个yaml： https ://github.com/ibm-messaging/message-hub-docs/blob/master/kafka-administration-api/KafkaTopicManagement.yaml

运行以下命令时出现 HTTP 405 Method Not Allowed： curl -X POST --tlsv1.2 -vk -H "Content-Type: application/vnd.kafka.binary.v1+json" -H "X-Auth -Token：来自我的应用程序环境变量的 apikey" https://kafka-rest-prod01.messagehub.services.us-south.bluemix.net/topics -d "{ \"TopicCreateParam\" { \"name\" : \"my.test\" }}"

谢谢你的帮助。

我在 Bluemix 上创建了一个新的 IBM Message Hub 实例，可以使用 Kafka Java 客户端（从我的笔记本电脑）连接到它并列出主题。我不能做的是通过管理 REST API 创建一个主题（为清楚起见添加了换行符）：

...这导致：

enable.auto.commit当我的消费者成功处理了一条记录时，我想告诉 Kafka，所以我通过设置为 false 来关闭自动提交。我在偏移量 0 和 1 处订阅了关于我订阅的主题的两条消息，并创建了一个使用者，以便每次调用poll最多返回一条记录（通过设置max.poll.records为 1）。

我现在打电话consumer.poll(5000)并收到第一条消息，但我不承认；我不打电话commitSync或commitAsync。如果我现在consumer.poll(5000)再次调用，使用相同的消费者，我希望得到与我刚刚阅读的完全相同的消息，但相反，我收到了第二条消息。

consumer.poll在我明确承认之前，我如何才能继续分发相同的消息？

根据 Bluemix，IBM Message hub 的计费基于 (a) “创建的分区数（每天计算）”和 (b) 发送的消息数。它说，创建的每个分区花费 10 美元。

在我们的例子中，我们有每天执行多次的单元测试，以确保我们创建和删除分区的代码正常工作。每次单元测试运行时，都会创建一个分区，然后将其删除。

如果我的单元测试每天运行 10 次，但删除了它创建的每个分区，我每天需要支付 100 美元吗？还是会因为我总是自己打扫而被收取 0 美元的费用？

当我们在 bluemix 中创建消息中心实例时，我们可以在屏幕上看到一个服务凭证选项卡。数据为 json 格式，不可编辑：

“用户”：“xxxxxxxxxxxxxxx”，“密码”：“xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx”

有什么方法可以自定义或编辑用户名和密码吗？

我正在尝试连接消息集线器和 apache kafka。我已经在 openwhisk 中创建了触发器，但我想知道我们是否可以在消息中心本身中创建触发器。

我们正在尝试实施 Kafka 作为我们的消息代理解决方案。我们正在 IBM BLuemix 中部署 Spring Boot 微服务，其内部消息代理实现是 Kafka 版本 0.10。由于我的经验更多是在 JMS、ActiveMQ 端，我想知道在 java 消费者中处理系统级错误的理想方法应该是什么？

以下是我们目前的实施方式

消费者财产

我们使用默认属性

卡夫卡消费者

我们为每个主题旋转 3 个线程，它们都具有相同的 groupId，即每个线程一个 KafkaConsumer 实例。到目前为止，我们只有一个分区。消费者代码在线程类的构造函数中是这样的

实际读取发生在线程的 run 方法中

您会注意到 EventProcessor 是一个处理每条记录的服务类，在大多数情况下提交数据库中的记录。如果处理器抛出错误（系统异常或验证异常），我们不会提交，而是以编程方式将搜索设置为该偏移量，以便后续轮询将从该组 id 的该偏移量返回。

现在的疑问是，这是正确的方法吗？如果我们遇到错误并且我们设置了偏移量，那么在修复之前不会处理其他消息。这可能适用于系统错误，例如无法连接到数据库，但如果问题仅在于该事件而不是其他事件来处理这一记录，我们将无法处理任何其他记录。我们想到了 ErrorTopic 的概念，当我们遇到错误时，消费者会将该事件发布到 ErrorTopic，同时它将继续处理其他后续事件。但是看起来我们正在尝试将 JMS 的设计概念（由于我之前的经验）引入 kafka，并且可能有更好的方法来解决 kafka 中的错误处理。此外，从错误主题重新处理它可能会更改某些情况下我们不想要的消息序列

请让我知道有人是如何按照 Kafka 标准在他们的项目中处理这种情况的。

-塔塔