在 Kafka Streams 中将所有消息延迟 30 分钟

Question

我们有一个用例，我们需要将主题 a 的所有消息写入主题 b，但每条消息延迟 30 分钟。你为什么问？由于时间对于这种数据流至关重要，因此付费客户可以获得实时馈送，对于免费加载者，我们提供延迟流。

我想在 KafkaConsumer poll() 循环中比较容易做到，通过比较系统时间和消息时间（使用有序的消息时间，如生产者时间或摄取时间），然后 pause()ing 有问题的分区并在最长 30 分钟的适当时间间隔后resume()他们（同时继续poll() 以避免故障转移）。

由于数据虽然有延迟，但仍需要流式传输，因此主题 a 和 b 中所有消息的摄取时间延迟应尽可能接近 30 分钟。

1. 但这在 Kafka Streams 中是否也很容易实现，以便我们可以使用其内置的完全一次性保证？我想知道"it's ok to call Thread.sleep() in Kafka Streams 是否也适用于长达 30 分钟的更长睡眠？ （当然我们不希望发生分区重新平衡，因为 Kafka 认为我们的流程有问题）

2. 假设我们让它工作，有没有办法对此进行适当的延迟监控？如果我们只是延迟消息，我认为消费者组的延迟总是至少 30 分钟的消息。那么是否可以让延迟监视器只计算超过 30 分钟的未处理消息？

（2. 对我们来说不如让 1. 工作重要）

编辑：https://stackoverflow.com/a/59261274/709537 提出了一个解决一些相关问题的方法，但它涉及到状态存储，因此看起来比我们简单的（？）“将所有消息延迟 x 分钟”任务看起来更复杂。

Answer 1

关于 2.，我认为我们必须为此推出自己的延迟监控。

做一些相关事情的简单方法 - 测量延迟而不是滞后消息的数量 - 将定期对每个分区进行

1. 获取第一条未读输入消息
2. currentLatency = max(0, ingestionTime(firstUnreadMessage) - 30min)

如果我们想监控滞后消息的数量，需要做一些更复杂的事情：

1. 向后读取输入消息，直到有ingestionTime + 30min <= systemTime
2. 这些消息的数量将是滞后

然而，向后阅读消息并不是 Kafka 的核心能力之一……可以设计一种巧妙的二分查找方式来获得准确的值。但是，没有人真正想知道消息延迟是 43123 还是 40513，他们想知道的是数量级。这将使搜索次数减少到少数（每个分区），并且不需要来回的二进制搜索样式。输出可以例如是

• 滞后
• 滞后
• 滞后
• 滞后
• ...