按键对Ktable分组后的时间戳无效（负）

Question

我正在使用KakfkaStreams(2.3.0)使用流和Ktable之间的连接来丰富一些值（压缩主题）。问题是压缩主题是使用与流主题不同的分区器编写的，因此连接没有按预期工作（一些键不匹配，因为它们在不同的分区中）。

我开始研究通过使用groupBy（）和reduce（）来重新分区压缩的主题，但当它开始读取创建的重新分区主题时，它开始抛出一个StreamsException并显示消息：

输入记录ConsumerRecord（topic=mappings-table-repartition,partition=18,leaderEpoch=null,offset=0,CreateTime=-1,serialized key size=37,serialized value size=20,headers=RecordHeaders（headers=[],isReadOnly=false）,key=0CECDec3863208E57,value=(1126999878035640323<-null))的时间戳无效（负）。可能是因为使用了0.10之前的producer客户机将该记录写入Kafka而没有嵌入时间戳，或者是因为输入主题是在将Kafka集群升级到0.10+之前创建的。使用不同的TimestampExtractor处理此数据。

重新分区是由KafkaStreams自动创建和填充的，所以这看起来很奇怪，它会写一个无效的记录。我甚至尝试在阅读最初的压缩主题时提供一个自定义的时间戳提取器，但它没有起到任何作用。stacktrace似乎表明它正在使用一个内部TimestampExtractor（查看代码确实如此）

在t org.apache.kafka.streams.processor.internals.streamthread.run（streamthread.java:774)

代码如下所示：

KTable<String, String> table = streamsBuilder
                .table(mappingsTopic,
                       Consumed.with(Serdes.String(), Serdes.String(),
                                      null,
                                      Topology.AutoOffsetReset.EARLIEST))
    .groupBy(KeyValue::pair,
             Grouped.with("mappings-table", Serdes.String(), Serdes.String()))
   .reduce((value1, value2) -> value2, (value1, value2) -> value2)

Answer 1

如错误消息所示，您需要升级代理使用的消息格式（参见.broker configlog.message.format.version)

从Kafka Streams1.0开始，需要0.10或更新的消息格式。

来源:https://kafka.apache.org/23/documentation/streams/upgrade-guide

Kafka Streams尝试在写入时设置记录时间戳，但是，旧的消息格式不支持时间戳（在写入时，当消息格式降级时，时间戳将丢失）。因此，在读取时，当消息格式转换回来时，将在消息中放入一个伪时间戳-1。

不允许为重新分区主题设置不同的时间戳提取器，因为上游时间戳必须向下游转发以确保正确性。