如何检测Kafka中KTable联表更新的触发方？

这个需求的核心就是要让联表输出的每条记录都带上触发源的标识，这样消费端就能一眼知道是Customers表还是Orders归约表的更新导致了新记录的生成。下面是具体的实现思路和步骤：

1. 先给两个源KTable加上触发标识 #

在把两个表拿来做关联之前，先给每个表的每条记录新增一个专门的字段（比如叫update_trigger），用来标记这条记录的更新类型：

• 对于Customers表：只要是客户信息（比如地址）更新，就给这条记录的update_trigger赋值为CUSTOMER_UPDATE；
• 对于归约后的OrdersAgg表（存储每个客户的订单总额）：每当订单总额发生变化（比如新增订单导致总额超过1000美元），就给这条记录的update_trigger赋值为ORDER_TOTAL_UPDATE。

举个Kafka Streams的代码示例：

// 处理Customers表，添加触发标识
KTable<String, CustomerWithTrigger> customersWithTrigger = customersTable
    .mapValues(customer -> {
        CustomerWithTrigger cwt = new CustomerWithTrigger();
        cwt.setCustomerId(customer.getId());
        cwt.setAddress(customer.getAddress());
        cwt.setUpdateTrigger("CUSTOMER_UPDATE");
        return cwt;
    });

// 处理订单归约表，添加触发标识
KTable<String, OrderAggWithTrigger> ordersAggWithTrigger = ordersAggTable
    .mapValues(orderAgg -> {
        OrderAggWithTrigger oawt = new OrderAggWithTrigger();
        oawt.setCustomerId(orderAgg.getCustomerId());
        oawt.setTotalAmount(orderAgg.getTotalAmount());
        oawt.setUpdateTrigger("ORDER_TOTAL_UPDATE");
        return oawt;
    });

2. 关联时保留并合并触发标识 #

在执行join操作的时候，把两个表的触发标识都带到结果里。如果遇到极端情况（两个表同一时间更新同一个客户的记录），还可以做个简单的合并逻辑，标记为BOTH_UPDATE：

KTable<String, JoinedCustomerOrder> joinedTable = customersWithTrigger
    .join(ordersAggWithTrigger,
        (customerTrigger, orderAggTrigger) -> {
            JoinedCustomerOrder result = new JoinedCustomerOrder();
            result.setCustomer(customerTrigger);
            result.setOrderAgg(orderAggTrigger);
            // 判断触发源
            if (customerTrigger.getUpdateTrigger().equals("CUSTOMER_UPDATE") 
                && orderAggTrigger.getUpdateTrigger().equals("ORDER_TOTAL_UPDATE")) {
                result.setTriggerSource("BOTH_UPDATE");
            } else {
                // 基于KTable的快照特性，取触发更新的一方标识
                result.setTriggerSource(
                    customerTrigger.getUpdateTrigger().equals("CUSTOMER_UPDATE") 
                    ? "CUSTOMER_UPDATE" 
                    : "ORDER_TOTAL_UPDATE"
                );
            }
            return result;
        }
    );

3. 消费端根据触发标识处理优惠逻辑 #

当消费joinedTable的输出时，就可以根据triggerSource字段来精准处理：

• 如果是CUSTOMER_UPDATE：检查客户的新地址是否在XYZ产品的售卖区域，符合条件就推送XYZ产品专属优惠；
• 如果是ORDER_TOTAL_UPDATE：检查订单总额是否超过1000美元，达标就推送满减或大额订单专属优惠；
• 如果是BOTH_UPDATE：可以选择同时推送两种优惠，或者根据业务优先级推送其中一种。

额外提示：处理首次关联的场景 #

如果是新客户第一次产生订单，这时候联表会生成第一条记录，两个源表的记录都是新增的。你可以在初始化的时候给触发标识设为INITIAL_CREATION，消费端可以对应推送新客户欢迎优惠。

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