KStream-KTable LeftJoin，在 KTable 未完全加载时发生 Join

Question

我正在尝试使用 KStream-KTable leftJoin 来丰富主题 A 和主题 B 的项目。主题 A 是我的 KStream，主题 B 是我的 KTtable，它有大约 23M 条记录。两个主题的键都没有计算，所以我必须使用 reducer 将 KStream(topic B) 转换为 KTable。

这是我的代码：

KTable<String, String> ktable = streamsBuilder
     .stream("TopicB", Consumed.withTimestampExtractor(new customTimestampsExtractor()))
     .filter((key, value) -> {...})
     .transform(new KeyTransformer()) // generate new key
     .groupByKey()
     .reduce((aggValue, newValue) -> {...});

streamBuilder
     .stream("TopicA")
     .filter((key, value) -> {...})
     .transform(...)
     .leftJoin(ktable, new ValueJoiner({...}))
     .transform(...)
     .to("result")

1) KTable 初始化很慢。 （大约 2000 msg/s），这是正常的吗？我的主题只有 1 个分区。有什么方法可以提高性能？ 我尝试设置以下内容以降低写入吞吐量，但似乎并没有太大改善。

CACHE_MAX_BYTES_BUFFERING_CONFIG = 10 * 1024 * 1024
COMMIT_INTERVAL_MS_CONFIG = 15 * 1000

2) 当从 Topic B 加载 KTable 时，连接发生。 这是连接发生时的偏移量（CURRENT-OFFSET/LOG-END-OFFSET）

   Topic A: 32725/32726 (Lag 1)
   Topic B: 1818686/23190390 (Lag 21371704)

我查看了Topic A失败记录的时间戳，是4天前的记录，Topic B最后一次处理的记录是6天前。 据我了解，kstream 进程记录基于时间戳，我不明白为什么在我的情况下，KStream(Topic A) 没有等到 KTable(Topic B) 完全加载到 4 天前触发加入。

我也尝试设置时间戳提取器返回 0，但效果不佳。

更新：将时间戳设置为 0 时，出现以下错误：

Caused by: org.apache.kafka.common.errors.UnknownProducerIdException: This exception is raised by the broker if it could not locate the producer metadata associated with the producerId in question. This could happen if, for instance, the producer's records were deleted because their retention time had elapsed. Once the last records of the producerID are removed, the producer's metadata is removed from the broker, and future appends by the producer will return this exception. 

我还尝试将 max.task.idle.ms 设置为 > 0（3 秒和 30 分钟），但仍然遇到相同的错误。

更新：我通过将 customTimestampsExtractor 设置为 6 天前修复了“UnknownProducerIdException”错误，这仍然早于主题 A 的记录。我认为（不确定）设置为 0 会触发导致此错误的更改日志上的保留。但是，在 ktable 完成加载之前，join 仍然无法正常工作。这是为什么呢？

我正在使用 Kafka Streams 2.3.0。

我在这里做错了吗？非常感谢。

Answer 1

1.KTable初始化慢。 （大约 2000 msg/s），这正常吗？

这取决于您的网络，我认为限制是TopicB的消耗率，您使用的两个配置CACHE_MAX_BYTES_BUFFERING_CONFIG和COMMIT_INTERVAL_MS_CONFIG是在您想要产生多少KTable输出之间进行权衡（因为 KTable 更改日志是修订流）以及当您将 KTable 更新到底层主题和下游处理器时您接受多少延迟。详细查看Kafka Streams caching config for state store 和this blog part Tables, Not Triggers。

我认为提高TopicB消费率的好方法是增加分区。

1. KStream.leftJoin(KTable,...) 始终是查表，总是将当前流记录与KTable 上最新更新的记录连接起来，在决定是否加入时不会考虑流时间。如果您想在加入时考虑直播时间，请查看KStream-KStream join。

在您的情况下，此滞后是 TopicB 的滞后，这并不意味着 KTable 未完全加载。当您的 KTable 处于状态恢复过程中时，当它从 KTable 的底层更改日志主题中读取以在实际运行您的流应用程序之前恢复当前状态时，您的 KTable 未完全加载，以防万一您将无法进行连接，因为流应用程序不是运行直到状态完全恢复。