KStream-KStream leftJoin窗口过期后输出延迟不一致的原因排查

我们有一项服务，支持用户订购太阳能板时搭配电池。在订单配置环节，我们会尝试获取电池产品详情，但即便获取失败，仍需将订单推送至CRM系统。为此我们用最新版Kafka Streams的leftJoin实现了以下流程：

• 监听order-received主题事件；
• 过滤掉不含电池产品的订单；
• 等待30分钟接收order-battery-details主题的事件；
• 若未收到电池详情事件，则将已有订单数据发送至battery-order主题。

当前遇到的问题是：当两个事件都收到时流程正常，但仅收到第一个事件时，输出不一致——有时30分钟窗口过期后立即输出，有时会延迟数小时。

我想了解：

1. 当窗口过期（未收到关联的右流事件）时，事件输出的触发条件是什么？
2. 导致长时间延迟的原因可能有哪些？

核心代码示例 #

@Component
class BatteryOrderProducer {
    @Autowired
    fun buildPipeline(streamsBuilder: StreamsBuilder) {
        // 监听新订单并过滤出包含电池的订单
        val orderReceivedReceivedStream = streamsBuilder.stream(
            "order-received",
            Consumed.with(Serdes.String(), JsonSerde<OrderReceivedEvent>())
        ).filter { _, order ->
            // 判断订单是否包含电池产品
        }.peek { key, order ->
            log.info("Received order with a battery product: $key", order)
        }

        // 监听电池详情事件
        val batteryDetailsStream = streamsBuilder
            .stream(
                "order-battery-details",
                Consumed.with(Serdes.String(), JsonSerde<BatteryDetailsEvent>())
            ).peek { key, order ->
                log.info("Received battery details: $key", order)
            }

        val valueJoiner: ValueJoiner<OrderReceivedEvent, BatteryDetailsEvent, BatteryOrder> =
            ValueJoiner { orderReceived: OrderReceivedEvent, batteryDetails: BatteryDetailsEvent? ->
                // 创建BatteryOrder对象
                if (batteryDetails != null) {
                    // 若获取到电池详情则添加至订单
                }
                // 返回BatteryOrder
            }

        // 无论是否收到第二个事件，都要推送电池订单
        orderReceivedReceivedStream.leftJoin(
            batteryDetailsStream,
            valueJoiner,
            JoinWindows.ofTimeDifferenceAndGrace(
                Duration.ofMinutes(30),
                Duration.ofMinutes(1)
            ),
            StreamJoined.with(
                Serdes.String(),
                JsonSerde<OrderReceivedEvent>(),
                JsonSerde<BatteryDetailsEvent>()
            ).withStoreName("battery-store")
        ).peek { key, value ->
            log.info("Merged BatteryOrder", value)
        }.to(
            "battery-order",
            Produced.with(
                Serdes.String(),
                JsonSerde<BatteryOrder>()
            )
        )
    }
}

问题解答 #

一、窗口过期时的输出触发条件 #

对于Kafka Streams的leftJoin窗口操作，未收到右流事件时，窗口过期后的输出触发依赖以下机制：

1. 窗口的实际可清理时间是主窗口时长（30分钟）加上宽限期（1分钟），即事件产生后31分钟，窗口进入可清理状态。
2. Kafka Streams的流任务会定期执行窗口清理任务（默认间隔30秒，可通过streams.cleanup.interval.ms配置），只有当清理任务运行时，才会检测到已过期的窗口，触发左流事件与null右值的join操作，将结果输出到目标主题。
3. 若流任务发生重启或重新平衡，在状态恢复阶段会扫描所有已过期的窗口，触发对应的输出操作。

二、导致长时间延迟的可能原因 #

• 清理任务间隔配置过大：如果streams.cleanup.interval.ms被设为几小时级别的值，清理任务很久才会运行一次，过期窗口无法及时处理。
• 流任务负载过高：当任务处理的事件量过大，或存在耗时的序列化/反序列化、复杂业务逻辑时，清理任务会被延迟执行，无法按时扫描过期窗口。
• 状态存储I/O瓶颈：窗口数据存储在RocksDB中（这里是battery-store），如果磁盘IO性能差、状态数据量过大，会拉长清理任务扫描过期窗口的时间，甚至引发阻塞。
• 宽限期内的重复事件干扰：宽限期内若有同key的重复order-received事件进入窗口，会重置窗口的过期时间，推迟输出时机。
• 频繁的流任务重新平衡：消费者组频繁触发重新平衡时，流任务会暂停处理，清理任务也会中断，直到重新平衡完成后才会继续执行，导致输出延迟。
• 时间戳不一致问题：Kafka Streams依赖事件时间戳计算窗口，若事件的CreateTime与流任务所在机器的系统时间偏差较大，会导致窗口过期时间计算错误，实际输出被推迟。

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