KStream-KTable Inner Join Lost Messages with Exactly Once Config答案

【问题标题】：KStream-KTable Inner Join Lost Messages with Exactly Once ConfigKStream-KTable Inner Join Lost Messages with Exactly Once Config
【发布时间】：2020-01-19 04:55:26
【问题描述】：

当我没有设置 processing.gurantee 时，这意味着流将以其默认值 (at_least_once) 启动，此代码可以成功记录并将加入的消息发送到相关话题。

当exactly_once配置在同一个流应用上启用时，部分数据无法通过join成功。即使有第一个 peek 块的日志，我也看不到一些第二个 peek 日志和一些我需要拥有的消息。

我确定 kstream 和 ktable 都需要不为空的值。并且双方都会定期收到消息。

流配置：

processing.guarantee=exactly_once
replication.factor=3（这会增加内部主题的复制因子）

Kafka（带 3 个代理）详细信息：

版本=2.2.0
log.roll.ms=3600000
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=3
message.max.bytes=2000024

问题是，exactly_once处理保证设置怎么会导致这种情况？

final KStream<String, UserProfile> userProfileStream = builder.stream(TOPIC_USER_PROFILE);
final KTable<String, Device> deviceKTable = builder.table(TOPIC_DEVICE);

userProfileStream
.peek((genericId, userProfile) ->
    log.debug("[{}] Processing user profile: {}", openUserId, userProfile)
)
.join(
    deviceKTable,
    (userProfile, device) -> {
        userProfile.setDevice(device);

        return userProfile;
    },
    Joined.with(Serdes.String(), userProfileSerde, deviceSerde)
)
.peek((genericId, userProfile) ->
    log.debug("[{}] Updated user profile: {}", genericId, userProfile)
)
.to(TOPIC_UPDATED_USER_PROFILE, Produced.with(Serdes.String(), userProfileSerde));

【问题讨论】：

标签： java apache-kafka apache-kafka-streams

【解决方案1】：

有关该问题的更多信息也在 Confluent 邮件组中讨论：https://groups.google.com/d/msg/confluent-platform/MRjz8MRBDCg/XbVlJI0hBAAJ

【讨论】：