如何在两个KTables上模拟右外键连接的最优方案？

针对你的需求——保留所有Rule记录（无论是否有对应的RuleAppliesTo）并关联其对应的人员信息，本质上是以Rule为主表的左外连接（对应你说的右外键连接），以下是优化后的解决方案：

核心问题分析 #

当前方案的两个关键缺陷：

• 聚合列表为空时未生成墓碑记录，导致下游无法感知RuleAppliesTo记录被完全删除的状态
• RuleAppliesTo的ruleId更新时，旧记录删除与新记录添加跨分区，处理顺序不确定，引发短暂数据不一致

优化方案 #

1. 修正聚合逻辑，正确处理墓碑记录 #

首先调整RuleAppliesTo的聚合代码，确保当某个ruleId对应的所有RuleAppliesTo被删除时，生成墓碑记录（返回null），这样下游左连接时能正确识别“无关联人员”的状态：

// 先定义List<RuleAppliesTo>的Serde，例如用JsonSerde或自定义Serde
Serde<List<RuleAppliesTo>> ruleAppliesToListSerde = Serdes.serdeFrom(
    new JsonSerializer<>(),
    new JsonDeserializer<>(new TypeReference<List<RuleAppliesTo>>() {})
);

KTable<String, List<RuleAppliesTo>> applicantsAgg = builder.table("applicant_topic")
    // 按ruleId重新分组，将RuleAppliesTo关联到对应的Rule
    .groupBy((applicantId, applicant) -> KeyValue.pair(applicant.getRuleId(), applicant))
    .aggregate(
        // 初始化空列表
        ArrayList::new,
        // 添加新的RuleAppliesTo记录
        (ruleId, newApplicant, currentList) -> {
            currentList.add(newApplicant);
            return currentList;
        },
        // 移除删除的RuleAppliesTo记录，列表为空时返回null生成墓碑
        (ruleId, deletedApplicant, currentList) -> {
            currentList.remove(deletedApplicant);
            return currentList.isEmpty() ? null : currentList;
        },
        // 指定状态存储和Serde，确保状态持久化和序列化正确
        Materialized.<String, List<RuleAppliesTo>, KeyValueStore<Bytes, byte[]>>as("applicants-agg-store")
            .withValueSerde(ruleAppliesToListSerde)
    );

2. 执行Rule与聚合表的左外连接 #

将Rule KTable与上述聚合表执行左外连接，确保所有Rule记录都被保留，无关联人员时用空集合填充：

// 定义结果对象，包含Rule和对应的人员列表
class RuleWithApplicants {
    private Rule rule;
    private List<RuleAppliesTo> applicants;

    // 构造方法、getter/setter省略
    public RuleWithApplicants(Rule rule, List<RuleAppliesTo> applicants) {
        this.rule = rule;
        this.applicants = applicants != null ? applicants : Collections.emptyList();
    }
}

// 执行左连接
KTable<String, RuleWithApplicants> ruleWithApplicants = ruleTable
    .leftJoin(applicantsAgg, RuleWithApplicants::new);

3. 处理ruleId更新的顺序问题 #

对于RuleAppliesTo的ruleId更新场景，CDC会发送旧记录的墓碑和新记录的插入（同一条RuleAppliesTo的id）：

• 由于Kafka Streams的最终一致性，即使旧记录删除和新记录添加跨分区导致短暂不一致，最终状态会自动修正
• 若业务无法接受短暂不一致，可以通过状态存储跟踪每个RuleAppliesTo的旧ruleId，确保先删除旧关联再添加新关联，示例代码如下（复杂度较高，按需选择）：

KStream<String, RuleAppliesTo> applicantStream = builder.stream("applicant_topic");

// 用状态存储跟踪每个RuleAppliesTo.id对应的旧ruleId
KStream<String, RuleAppliesTo> trackedStream = applicantStream.transformValues(
    () -> new ValueTransformerWithKey<String, RuleAppliesTo, RuleAppliesTo>() {
        private KeyValueStore<String, String> oldRuleIdStore;

        @Override
        public void init(ProcessorContext context) {
            oldRuleIdStore = (KeyValueStore<String, String>) context.getStateStore("old-rule-id-store");
        }

        @Override
        public RuleAppliesTo transform(String applicantId, RuleAppliesTo applicant) {
            if (applicant == null) {
                // 处理墓碑记录：删除旧ruleId的关联
                String oldRuleId = oldRuleIdStore.get(applicantId);
                if (oldRuleId != null) {
                    context.forward(oldRuleId, null, To.child("delete-old"));
                    oldRuleIdStore.delete(applicantId);
                }
                return null;
            } else {
                String oldRuleId = oldRuleIdStore.get(applicantId);
                // 如果ruleId变更，先删除旧关联
                if (oldRuleId != null && !oldRuleId.equals(applicant.getRuleId())) {
                    context.forward(oldRuleId, applicant, To.child("delete-old"));
                }
                oldRuleIdStore.put(applicantId, applicant.getRuleId());
                return applicant;
            }
        }

        @Override
        public void close() {}
    },
    "old-rule-id-store" // 提前在Topology中注册该状态存储
);

// 合并删除旧关联的流与新增关联的流
KStream<String, RuleAppliesTo> combinedStream = trackedStream
    .merge(trackedStream.selectKey((k, v) -> v != null ? v.getRuleId() : k))
    .merge(trackedStream.selectKey((k, v) -> k).filter((k, v) -> v == null));

// 再执行聚合操作（同步骤1的聚合逻辑）
KTable<String, List<RuleAppliesTo>> applicantsAgg = combinedStream
    .groupByKey()
    .aggregate(/* 同步骤1的聚合逻辑 */);

关键注意事项 #

• 确保Rule和RuleAppliesTo的Serde实现正确，尤其是列表类型的Serde
• 状态存储的配置要符合业务的持久化需求，避免数据丢失
• 若使用CDC主题，确保主题的压缩策略（如cleanup.policy=compact）正确配置，以优化存储和处理效率

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