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

Question

我正在尝试使用KStream-KTable leftJoin来丰富主题A中的条目和主题B。主题A是我的KStream，主题B是我的KTtable，它有大约2300万条记录。这两个主题中的键都没有计算，所以我必须使用reducer将KStream（主题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）当KTable未从主题B加载完成时发生连接。这里是连接发生时的偏移量（current-offset/log-end-offset）

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

更新：将时间戳设置为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天前（仍早于主题A中的记录）修复了“UnknownProducerIdeXception”错误。我将thhink（不确定）设置为0触发了导致此错误的changelog上的保留。但是，在ktable完成加载之前，如果join仍然发生，则join仍然不起作用。这是为什么？

我使用的是Kafka Streams 2.3.0。

我在这里做错什么了吗？多谢。

Answer 1

1.KTable初始化速度慢。（2000味精/秒左右），这正常吗？

这取决于您的网络，我认为限制是TopicB的消耗率，您使用的两个configcache_max_bytes_buffering_config和commit_interval_ms_config是为了在您希望生成多少KTable输出（因为KTable changelog是修订流）和您在将KTable更新到底层主题和下游处理器时接受多少延迟之间进行权衡。详细查看状态存储的Kafka流缓存配置和此blog部分表，而不是触发器。

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

在您的示例中，此滞后是topicb的滞后，它并不意味着KTable未完全加载。当您的KTable处于state restore过程中时，它没有被完全加载，当它在实际运行您的stream app之前从KTable的底层changelog主题读取以恢复当前状态时，您就无法进行连接，因为stream app在state完全恢复之前才运行。