如何为Kafka Streams消费重试主题消息添加处理延迟？

我懂你现在的痛点——用Kafka Streams处理重试主题的消息，还得加上延迟机制，避免频繁重试把系统压垮，而且绝对不想用sleep那种阻塞线程的粗暴做法。结合你给出的简化代码，我给你几个优雅的实现方案，覆盖你提到的几种需求场景，你可以按需挑选：

方案一：每条消息处理后延迟固定时间再处理下一条（对应需求4） #

这个方案适合单条消息慢处理的场景，用Kafka Streams的process API结合异步延迟队列实现，完全不会阻塞Streams的工作线程：

import org.apache.kafka.streams.processor.api.Processor
import org.apache.kafka.streams.processor.api.ProcessorContext
import org.apache.kafka.streams.processor.api.Record
import java.util.concurrent.CompletableFuture
import java.util.concurrent.TimeUnit

fun createTopology(topic: String): Topology {
    val DELAY_SECONDS = 5L
    val streamsBuilder = StreamsBuilder()
    streamsBuilder.stream<String, ArchivalData>(topic, Consumed.with(Serdes.String(), ArchivalDataSerde()))
        .peek { key, msg -> logger.info("Received event for key $key : $msg") }
        .process({
            object : Processor<String, ArchivalData, String, ArchivalData> {
                private lateinit var context: ProcessorContext<String, ArchivalData>

                override fun init(context: ProcessorContext<String, ArchivalData>) {
                    this.context = context
                }

                override fun process(record: Record<String, ArchivalData>) {
                    // 异步延迟处理，避免阻塞Streams工作线程
                    CompletableFuture.delayedExecutor(DELAY_SECONDS, TimeUnit.SECONDS)
                        .execute {
                            try {
                                val enrichedMsg = enrich(record.value())
                                archive(enrichedMsg)
                                // 处理成功，提交偏移量
                                context.commit()
                            } catch (e: Exception) {
                                logger.error("Failed to process event ${record.value()}, putting back to retry topic", e)
                                // 处理失败，将消息转发回原重试主题
                                context.forward(record)
                                context.commit()
                            }
                        }
                }

                override fun close() {}
            }
        })
    return streamsBuilder.build()
}

说明：通过CompletableFuture.delayedExecutor实现异步延迟，Streams的工作线程可以继续处理其他任务，不会被阻塞。处理成功就提交偏移量，失败则将消息转发回原主题实现重试，建议搭配processing.guarantee=AT_LEAST_ONCE配置避免消息丢失。

方案二：只处理“存入主题超过X秒”的消息（对应需求3） #

这个方案利用Kafka消息的时间戳，过滤出已经在主题中停留足够时间的消息再处理，适合希望消息在重试队列中“冷却”一段时间的场景：

fun createTopology(topic: String): Topology {
    val DELAY_SECONDS = 5L
    val streamsBuilder = StreamsBuilder()
    streamsBuilder.stream<String, ArchivalData>(topic, Consumed.with(Serdes.String(), ArchivalDataSerde()))
        .peek { key, msg -> logger.info("Received event for key $key : $msg") }
        // 过滤出存入时间超过指定延迟的消息
        .filter { _, _, recordContext ->
            val currentTime = System.currentTimeMillis()
            val recordTimestamp = recordContext.timestamp()
            currentTime - recordTimestamp >= DELAY_SECONDS * 1000
        }
        .map { key, msg -> enrich(msg) }
        .foreach { key, enrichedMsg ->
            try {
                archive(enrichedMsg)
            } catch (e: Exception) {
                logger.error("Failed to process event $enrichedMsg, putting back to retry topic", e)
                // 处理失败，手动将消息发回重试主题
                retryProducer.send(ProducerRecord(topic, key, enrichedMsg))
            }
        }
    return streamsBuilder.build()
}

说明：默认情况下Kafka消息的时间戳是生产者发送时间，如果是手动重试的消息，要确保重新发送时更新时间戳，这样延迟逻辑才会生效。处理失败后需要手动将消息发送回重试主题，因为过滤后的流不会自动回退消息。

方案三：批量处理后延迟固定时间再继续消费（对应需求2） #

这个方案适合批量处理场景，用punctuate API定时触发批量处理，控制消费节奏：

import org.apache.kafka.streams.processor.api.Processor
import org.apache.kafka.streams.processor.api.ProcessorContext
import org.apache.kafka.streams.processor.api.Record
import java.util.concurrent.TimeUnit

fun createTopology(topic: String): Topology {
    val BATCH_DELAY_SECONDS = 10L
    val streamsBuilder = StreamsBuilder()
    streamsBuilder.stream<String, ArchivalData>(topic, Consumed.with(Serdes.String(), ArchivalDataSerde()))
        .process({
            object : Processor<String, ArchivalData, String, ArchivalData> {
                private lateinit var context: ProcessorContext<String, ArchivalData>
                private val batch = mutableListOf<Record<String, ArchivalData>>()

                override fun init(context: ProcessorContext<String, ArchivalData>) {
                    this.context = context
                    // 每隔指定时间触发一次批量处理
                    context.schedule(BATCH_DELAY_SECONDS, TimeUnit.SECONDS) {
                        if (batch.isNotEmpty()) {
                            processBatch(batch)
                            batch.clear()
                            context.commit()
                        }
                    }
                }

                override fun process(record: Record<String, ArchivalData>) {
                    batch.add(record)
                }

                private fun processBatch(batch: List<Record<String, ArchivalData>>) {
                    batch.forEach { record ->
                        try {
                            val enrichedMsg = enrich(record.value())
                            archive(enrichedMsg)
                        } catch (e: Exception) {
                            logger.error("Failed to process event ${record.value()}, putting back to retry topic", e)
                            context.forward(record)
                        }
                    }
                }

                override fun close() {}
            }
        })
    return streamsBuilder.build()
}

说明：通过schedule设置定时任务，每隔指定时间处理一次累积的消息批次，平衡处理效率和系统压力。你可以根据业务调整批次大小和延迟时间，避免短时间内处理过多消息。

内容的提问来源于stack exchange，提问作者sjoblomj