Kafka Streams一对多外键关联 重发布新闻如何覆盖历史关联图片

当前实现将每条新闻事件拆分为多条单图片ID的记录做关联，聚合逻辑仅做图片追加操作，没有感知到同新闻ID的更新事件，无法清除该新闻之前关联的旧图片记录，因此会累积所有历史关联的图片，不符合更新覆盖的需求。

方案一：使用GlobalKTable直接组装（推荐，无重分区开销） #

如果图片总数据量可以被单节点承载，直接将images Topic构建为全局表，每条新闻事件直接查询全局表组装结果，天然支持更新覆盖：

// 构建全局图片表，所有实例存储全量图片数据
GlobalKTable<String, Image> images = builder.globalTable(topics.getImagesTopic(), Consumed.with(Serdes.String(), imageSerde));
KStream<String, News> news = builder.stream(topics.getNewsTopic(), Consumed.with(Serdes.String(), newsSerde));

news.mapValues(newsArticle -> {
    NewsImages result = new NewsImages();
    result.setNewsId(newsArticle.getId());
    result.setTitle(newsArticle.getTitle());
    result.setDescription(newsArticle.getDescription());
    result.setAuthor(newsArticle.getAuthor());
    Map<String, Image> imageMap = new HashMap<>();
    // 直接从全局表查询当前新闻关联的所有图片
    for (String imageId : newsArticle.getImageIds()) {
        Image image = images.get(imageId);
        if (image != null) {
            imageMap.put(imageId, image);
        }
    }
    result.setImages(imageMap);
    return result;
}).to(topics.getNewsImagesTopic(), Produced.with(Serdes.String(), newsImagesSerde));

该方案无需重分区、无需聚合，逻辑简单，每次新闻更新都会重新拉取当前关联的图片，自然覆盖旧结果，不会出现历史图片累积问题。

方案二：改造现有聚合逻辑（适合图片数据量大无法用全局表的场景） #

给News类新增version字段，每次新闻更新时version值递增，改造聚合逻辑做版本判断：

1. 调整ImageWrapper类，新增version字段存储对应新闻的版本号
2. 调整NewsImages类，新增version字段存储当前聚合结果对应的新闻版本
3. 改造聚合逻辑如下：

KTable<String, Image> images = builder.table(topics.getImagesTopic(), Consumed.with(Serdes.String(), imageSerde));
KStream<String, News> news = builder.stream(topics.getNewsTopic(), Consumed.with(Serdes.String(), newsSerde));

KTable<String, NewsImages> newsImagesKTable = news.flatMapValues(newsArticle -> 
        // 拆分时携带新闻版本
        newsArticle.getImageIds().stream().map(id -> new ImageIdWithVersion(id, newsArticle.getVersion())).collect(Collectors.toList())
    )
    .map((newsId, idWithVersion) -> new KeyValue<>(idWithVersion.getImageId(), new NewsIdWithVersion(newsId, idWithVersion.getVersion())))
    .join(images, (newsIdWithVersion, image) -> new ImageWrapper(newsIdWithVersion.getNewsId(), image, newsIdWithVersion.getVersion()), 
        Joined.with(Serdes.String(), newsIdWithVersionSerde, imageSerde))
    .groupBy((imageId, imageWrapper) -> imageWrapper.getNewsId(), Grouped.with(Serdes.String(), imageWrapperSerde))
    .aggregate(NewsImages::new, (newsId, imageWrapper, newsImages) -> {
        // 版本更新则清空旧图片
        if (imageWrapper.getVersion() > newsImages.getVersion()) {
            newsImages.getImages().clear();
            newsImages.setVersion(imageWrapper.getVersion());
            newsImages.setNewsId(newsId);
        }
        // 同版本的图片正常追加
        newsImages.addImage(imageWrapper.getImage());
        return newsImages;
    }, Materialized.with(Serdes.String(),newsImagesSerde));

newsImagesKTable.toStream().to(topics.getNewsImagesTopic());

如果存在新闻发布时对应图片还未写入images Topic的场景，可以给新闻流添加小窗口等待，或者使用left join处理缺失图片的兜底逻辑。

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