Kafka Streams：用于聚合的自定义时间戳提取器

Question

我正在构建一个非常简单的KafkaStreams演示应用程序，以测试用例。

我无法升级我正在使用的Kafka broker（当前版本为0.10.0），并且有几条消息是由0.10.0之前的生产者编写的，因此我使用自定义时间戳提取器，我将其作为默认值添加到主类开头的配置中：

config.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, GenericRecordTimestampExtractor.class);

当从我的源主题消费时，这非常好。但是当使用聚合运算符时，我遇到了一个异常，因为当从内部聚合主题消费时，使用了TimestampExtractor的FailOnInvalidTimestamp实现而不是自定义实现。

Streams应用程序的代码如下所示：

...

KStream<String, MyValueClass> clickStream = streamsBuilder
              .stream("mytopic", Consumed.with(Serdes.String(), valueClassSerde));

KTable<Windowed<Long>, Long> clicksByCustomerId = clickStream
              .map(((key, value) -> new KeyValue<>(value.getId(), value)))
              .groupByKey(Serialized.with(Serdes.Long(), valueClassSerde))
              .windowedBy(TimeWindows.of(TimeUnit.MINUTES.toMillis(1)))
              .count();
...

我遇到的例外情况如下：

    Exception in thread "click-aggregator-b9d77f2e-0263-4fa3-bec4-e48d4d6602ab-StreamThread-1" org.apache.kafka.streams.errors.StreamsException: 
Input record ConsumerRecord(topic = click-aggregator-KSTREAM-AGGREGATE-STATE-STORE-0000000002-repartition, partition = 9, offset = 0, CreateTime = -1, serialized key size = 8, serialized value size = 652, headers = RecordHeaders(headers = [], isReadOnly = false), key = 11230, value = org.example.MyValueClass@2a3f2ea2) has invalid (negative) timestamp. 
Possibly because a pre-0.10 producer client was used to write this record to Kafka without embedding a timestamp, or because the input topic was created before upgrading the Kafka cluster to 0.10+. Use a different TimestampExtractor to process this data.

现在的问题是：有什么方法可以让Kafka Streams在从内部聚合主题读取时使用自定义的TimestampExtractor（最佳情况下仍然使用Streams DSL）？

Answer 1

读完马蒂亚斯的回答后，我仔细检查了一切，问题的原因是Kafka经纪人和Kafka流应用程序之间的不兼容版本。 <罢工> 我愚蠢到使用Kafka Streams 1.0.0与0.10.1.1Broker，这在Kafka Wiki中被明确声明为不兼容 这里。

编辑（thx to Matthias）：问题的实际原因是我们的0.10.1使用的日志格式。x broker仍然是0.9.0。x、 与Kafka流不兼容。

Answer 2

这是众所周知的问题：-）。我对项目中的老客户机也有同样的问题，这些客户机仍然使用老Kafka客户机，比如0.9，并且在与一些“未经认证”的客户机通信时也存在同样的问题。网络客户端。

所以我写了一篇专题文章：

public class MyTimestampExtractor implements TimestampExtractor {

    private static final Logger LOG = LogManager.getLogger( MyTimestampExtractor.class );

    @Override
    public long extract ( ConsumerRecord<Object, Object> consumerRecord, long previousTimestamp ) {
        final long timestamp = consumerRecord.timestamp();

        if ( timestamp < 0 ) {
            final String msg = consumerRecord.toString().trim();
            LOG.warn( "Record has wrong Kafka timestamp: {}. It will be patched with local timestamp. Details: {}", timestamp, msg );
            return System.currentTimeMillis();
        }

        return timestamp;
    }
}

当有许多消息时，您可能会跳过日志记录，因为它可能会泛滥。

Answer 3

您不能更改时间戳提取器（如v1.0.0）。出于正确性的原因，这是不允许的。

但我真的很想知道，一开始是如何将时间戳为-1的记录写入这个主题的。Kafka Streams在写入记录时使用自定义提取器提供的时间戳。还要注意，KafkaProducer不允许写入时间戳为负的记录。

因此，我能想到的唯一解释是，其他制作人确实在重新划分主题中写了文章——这是不允许的。。。只有Kafka流应该写入重新分区主题。

我想，你需要删除这个主题，让Kafka流重新创建它，回到一个干净的状态。

根据对另一个答案的讨论/评论：

使用Kafka Streams需要0.10格式。如果升级代理并保持0.9或更旧的格式，Kafka Streams可能无法按预期工作。