Kafka流时间戳提取器在KGroupedStream处失败

Question

这个服务我已经测试了它，使用不同版本的Kafka（更高或等于0.10)，它工作良好。

以下是我的配置：

Spring：cloud：stream：kafka：streams：binder：brokers：${KAFKA_BROKERS}applicationid：email-MESSAGES-stream configuration：default.key.serde：org.apache.kafka.common.serialization.serdes$stringserde default.value.serde：org.apache.kafka.common.serialization.serdes$stringserde commit.interval.ms:1000 default.timestamp.extractor：

我代码的一部分：

    stream
        .mapValues(this::mapMessage)
        .groupBy(this::buildGroup, Serialized.with(new JsonSerde<>(Group.class), new JsonSerde<>(EmailMessage.class)))
        .windowedBy(TimeWindows.of(WINDOW_TIME))
        .aggregate(ArrayList::new, this::aggregate, Materialized.with(new JsonSerde<>(Group.class), new MessageListSerialization()))
        .toStream()
        .process(() -> new MailMessagesProcessor(emailService));

它向我抛出了这个错误：org.apache.kafka.streams.errors.streamsException：输入记录ConsumerRecord（主题=.....）使用不同的TimestampExtractor处理此数据。

Answer 1

Kafka Streams需要brokers 0.10.0或更新版本。它不适合较老的经纪人。

>

Kafka Streams 0.10.0仅与0.10.0（或更新）代理兼容。

Kafka Streams 0.10.1和更新版本，向后兼容0.10.1（但不兼容较旧的代理），并兼容较新的代理。

此外，由于Kafka Streams1.0，因此需要0.10（或更高）的消息格式。因此，即使您将代理升级到0.10.0（或更高），如果您的消息格式没有升级，它也不会工作。

为了使用“精确一次”特性，需要0.11.0（或更高）的broker版本。

有关更多详细信息，请参阅:https://docs.confluent.io/current/streams/upgrade-guide.html#兼容性