延迟 Kafka Streams 消费

Question

我正在尝试使用 Kafka Streams（即不是简单的 Kafka 消费者）从重试主题中读取之前无法处理的事件。我希望从重试主题中消费，如果处理仍然失败（例如，如果外部系统关闭），我希望将事件放回重试主题。因此，我不想立即继续消费，而是在消费前等待一段时间，以免系统中充斥着暂时无法处理的消息。

简化，目前的代码是这样做的，我想给它加一个延迟。

fun createTopology(topic: String): Topology {
    val streamsBuilder = StreamsBuilder()

    streamsBuilder.stream<String, ArchivalData>(topic, Consumed.with(Serdes.String(), ArchivalDataSerde()))
        .peek { key, msg -> logger.info("Received event for key $key : $msg") }
        .map { key, msg -> enrich(msg) }
        .foreach { key, enrichedMsg -> archive(enrichedMsg) }

    return streamsBuilder.build()
}

我曾尝试使用 Window Delay 进行设置，但未能成功。我当然可以在 peek 内睡一觉，但这会导致线程挂起，听起来不是一个非常干净的解决方案。

延迟如何工作的确切细节对我的用例来说并不是非常重要。例如，所有这些都可以正常工作：

1. 过去x 秒内关于该主题的所有事件都被一次性消耗掉。在开始/结束消费后，流在再次消费前等待 x 秒
2. 每个事件在被放到主题后x 秒处理
3. 流使用消息，每个事件之间的延迟为 x 秒

如果有人能提供几行 Kotlin 或 Java 代码来完成上述任何操作，我将不胜感激。

Answer 1

您不能真正暂停使用 Kafka Streams 从输入主题中读取——“延迟”的唯一方法是调用“睡眠”，但正如您所提到的，这会阻塞整个线程并且不是一个好的解决方案。

但是，您可以做的是使用有状态处理器，例如，process()（带有附加的状态存储）而不是foreach()。如果重试失败，您不会将记录放回输入主题，而是将其放入存储中并注册一个带有所需重试延迟的标点符号。如果标点触发，则重试，如果重试成功，则从存储中删除条目并取消标点；否则，请等到标点符号再次触发。