如何处理KStream与KTable关联时流元素键存于表记录中的场景

类结构 #

Order {String customerId, String orderId, List<Product> products}
Product {String id, String name, int quantity}
DeliveryNote {String id (orderId + productId + customerId), String orderId, int quantity}

当前订单流处理逻辑 #

KStream<String, Order> orderStream = StreamBuilder.toStream("orders");

KTable<String, DeliveryNote> deliveryNoteTable =
    orderStream
        .mapValues((orderKey, order) -> order to List<DeliveryNote>)
        .flatMap(DeliveryNotes.stream())
        .selectKey(DeliveryNote::id)
        .groupByKey()
        .aggregate(DeliveryNote::new, (deliveryNoteId, existingNote, newNote) -> existingNote.apply(newNote));

需求与痛点 #

需要将orderStream与deliveryNoteTable关联，当Order中无Product时，关闭该订单对应的所有DeliveryNote。但Order对象无法提供所有关联DeliveryNote的ID，而deliveryNoteTable中的所有记录都包含关联的orderId，但外键提取器仅支持从deliveryNoteTable关联订单KTable的场景，无法反向扫描表中匹配orderId的记录。

方案1：维护基于orderId的关联表 #

通过额外生成一张以orderId为key、对应DeliveryNote列表为value的KTable，实现快速关联：

1. 在现有deliveryNoteTable的生成逻辑外，新增关联表的构建：

KTable<String, List<DeliveryNote>> orderToDeliveryNotesTable =
    orderStream
        .mapValues((orderKey, order) -> order to List<DeliveryNote>)
        .flatMap(DeliveryNotes.stream())
        .groupBy((key, note) -> note.getOrderId()) // 按orderId分组聚合
        .aggregate(
            ArrayList::new,
            (orderId, note, noteList) -> {
                noteList.add(note);
                return noteList;
            },
            (orderId, note, noteList) -> {
                noteList.remove(note);
                return noteList;
            }
        );

2. 处理无Product的订单，关联获取对应的DeliveryNote并更新状态：

orderStream
    .filter((key, order) -> order.getProducts() == null || order.getProducts().isEmpty())
    .leftJoin(orderToDeliveryNotesTable, (targetOrder, relatedNotes) -> relatedNotes)
    .flatMapValues(notes -> notes)
    .mapValues(note -> {
        // 假设DeliveryNote有状态字段，标记为关闭
        note.setStatus("CLOSED");
        return note;
    })
    .selectKey(DeliveryNote::getId)
    .toTable()
    // 将更新合并到原deliveryNoteTable
    .join(deliveryNoteTable, (updatedNote, originalNote) -> updatedNote)
    .toStream()
    .to("delivery-notes-updates");

方案2：使用交互式查询（Interactive Queries） #

适合非实时或低频次场景，通过API查询状态存储：

1. 当无Product的订单到达时，调用Kafka Streams的交互式查询API，获取deliveryNoteTable的所有分区信息。
2. 遍历每个分区，查询该分区内所有orderId匹配的DeliveryNote记录。
3. 生成关闭这些DeliveryNote的消息，发送到对应Kafka主题，让deliveryNoteTable消费更新。
   注意：这种方式会有一定性能开销，不适合高并发场景。

方案3：调整DeliveryNote的Key结构（需业务允许） #

如果可以修改DeliveryNote的Key格式，将orderId作为Key的前缀（比如{orderId}-{customerId}-{productId}），就可以利用RocksDB状态存储的范围查询能力，直接获取某个orderId下的所有DeliveryNote。但这种调整会影响现有业务的Key依赖，需谨慎评估。

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