Pulsar 中的意外积压大小

Posted 2023-02-18

【中文标题】Pulsar 中的意外积压大小

【英文标题】：Unexpected backlog size in Pulsar

【发布时间】：2022-01-06 10:40:08

【问题描述】：

我正在使用 Pulsar 进行服务之间的通信，但我在对生产者和消费者的一个非常简单的测试中遇到了问题。

在 JUnit 4 测试中，我启动（我自己的包装器）一个 ZooKeeper 服务器、一个 BookKeeper bookie 和一个 PulsarService；配置应该很标准。

测试可以概括为以下步骤：

建立一个生产者； 建立消费者（例如，Pulsar 主题的读者）； 检查message backlog（使用精确积压）； 这是通过PulsarAdmin#topics#getStats#subscriptions获取当前订阅来完成的 我希望它是 0，因为没有发送关于该主题的任何内容，但有时它是 1，但这似乎是另一个问题... 构建一个新的生产者并同步向主题发送消息； 建立一个新的消费者并阅读该主题的消息； 我预计一封邮件会积压，但我实际上阅读了一封 新建生产者，同步发送4条消息； 再次获取消息，使用在步骤 5 中读取的 messageID 作为start message ID； 我预计这里会积压 4 条消息，并且大多数情况下这个值是正确的，但运行测试大约 10 次我始终得到 2 或 5 条

我尝试调试测试，但无法弄清楚这些值的来源；我是不是误会了什么？

【参考方案1】：

如果尚未完成，您可以尝试的事情：

要求进行精确的积压测量。默认情况下，它仅被估计为获得精确测量是一项成本更高的操作。为此使用admin.topics().getStats(topic, true)。 （见https://github.com/apache/pulsar/blob/724523f3051def9577d6bd27697866c99f4a7b0e/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java#L862） 在生产者端停用批处理。 msgBacklog 中返回的数字是条目数，因此在单个条目中批处理的多条消息将计为 1。请参阅相关问题：https://github.com/apache/pulsar/issues/7623。它可以解释如果 4 条消息已放入同一批次，为什么您会看到 msgBacklog 的值为 2。请注意，停用批处理会对性能产生巨大影响。

【讨论】：

是的，我已经在使用精确的 backlog（我会更新添加这个的问题，这实际上是一个重要的细节，谢谢）。说到批处理，我不确定，我肯定会检查，但我对此表示怀疑：如果它返回批处理的数量，它应该总是返回 2，而不是有时返回 2，有时是 5，大多数时候是 4。你同意我的观点吗？

嗨，我还尝试在生产者端禁用批处理，现在不稳定的测试不再不稳定了：事实上，它始终失败，积压为 5 而不是 4。

我认为我们设法在该测试中找到了问题：因为启用了批处理，并且我们在 Pulsar 上的抽象层总是利用异步发送在代理上发布，消息可能以不同的方式批处理，并且我们经历的值可以解释如下： - 2：所有消息都被批处理，除了第一个中间有消费操作的消息（所以 1，[2-5]）； - 4：只有最后两条消息（时间最近）被批处理（所以 1,2,3,[4-5]）； - 5：没有消息被批处理。

另外，由于我们的抽象层是如何工作的（它不发送确认，但它存储要读取的最后一个消息 ID）以及由于积压消息计数的解决方式，这很有意义在禁用批处理时获得值 5：管理实例没有提供开始读取的消息，因此积压计算测试中产生的所有消息。 @ChristopheBornet 对你有意义吗？