apache-kafka - Kafka 流在 StreamTask 中解码时间戳元数据失败

Question

我们在启动应用程序时在 Kafka Streams 上遇到了奇怪的错误

java.lang.IllegalArgumentException: Illegal base64 character 7b
    at java.base/java.util.Base64$Decoder.decode0(Base64.java:743)
    at java.base/java.util.Base64$Decoder.decode(Base64.java:535)
    at java.base/java.util.Base64$Decoder.decode(Base64.java:558)
    at org.apache.kafka.streams.processor.internals.StreamTask.decodeTimestamp(StreamTask.java:985)
    at org.apache.kafka.streams.processor.internals.StreamTask.initializeTaskTime(StreamTask.java:303)
    at org.apache.kafka.streams.processor.internals.StreamTask.initializeMetadata(StreamTask.java:265)
    at org.apache.kafka.streams.processor.internals.AssignedTasks.initializeNewTasks(AssignedTasks.java:71)
    at org.apache.kafka.streams.processor.internals.TaskManager.updateNewAndRestoringTasks(TaskManager.java:385)
    at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:769)
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:698)
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:671)

因此，有关失败流的错误:错误 KafkaStreams - stream-client [xxx] 所有流线程都已死亡。该实例将处于错误状态，应关闭。

根据 org.apache.kafka.streams.processor.internals.StreamTask 中的代码，失败是由于解码时间戳元数据时出错 (StreamTask.decodeTimestamp() ).它发生在产品上，无法在舞台上重现。此类错误的根本原因可能是什么？

额外信息:我们的应用程序使用 Kafka-Streams 并使用相同的 application.id 和 state.dir 使用来自多个 kafka broker 的消息(实际上我们从一个 broker 切换到另一个，但在某个时期我们连接到两个经纪人，所以我们有两个 kafka 流，每个经纪人一个)。据我了解，消费者群体生活在经纪人方面(所以应该不是问题)，但状态目录在客户端。也许由于对两个 kafka 流使用相同的 state.dir 而发生了一些竞争条件？这可能是根本原因吗？

我们使用kafka-streams v.2.4.0, kafka-clients v.2.4.0 , Kafka Broker v.1.1.1，配置如下:

default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
default.value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
default.timestamp.extractor: org.apache.kafka.streams.processor.WallclockTimestampExtractor
default.deserialization.exception.handler: org.apache.kafka.streams.errors.LogAndContinueExceptionHandler
commit.interval.ms: 5000
num.stream.threads: 1
auto.offset.reset: latest

Answer 1

最后，我们找出了某些消费者群体元数据损坏的根本原因。它是我们的内部监控工具之一(用 pykafka 编写)，暂时不活跃的消费者群体破坏了元数据。元数据未加密且包含无效数据，如下所示:{"consumer_id": "", "hostname": "monitoring-xxx"} .为了了解我们在消费者元数据中到底有什么，我们可以使用以下代码:

Map<String, Object> config = Map.of( "group.id", "...", "bootstrap.servers", "...");
String topicName = "...";
Consumer<byte[], byte[]> kafkaConsumer = new KafkaConsumer<byte[], byte[]>(config, new ByteArrayDeserializer(), new ByteArrayDeserializer());
Set<TopicPartition> topicPartitions = kafkaConsumer.partitionsFor(topicName).stream()
        .map(partitionInfo -> new TopicPartition(topicName, partitionInfo.partition()))
        .collect(Collectors.toSet());
kafkaConsumer.committed(topicPartitions).forEach((key, value) ->
    System.out.println("Partition: " + key + " metadata: " + (value != null ? value.metadata() : null)));

修复已损坏的元数据的几个选项:

• 将消费者组更改为新的消费者组。请注意，您可能会丢失或重复消息，具体取决于 latest或 earliest偏移重置策略。所以在某些情况下，这个选项可能是 Not Acceptable

• 手动覆盖元数据(时间戳根据内部逻辑编码 StreamTask.decodeTimestamp() ):

  Map<TopicPartition, OffsetAndMetadata> updatedTopicPartitionToOffsetMetadataMap = kafkaConsumer.committed(topicPartitions).entrySet().stream()
.collect(Collectors.toMap(Map.Entry::getKey, (entry) -> new OffsetAndMetadata((entry.getValue()).offset(), "AQAAAXGhcf01")));
kafkaConsumer.commitSync(updatedTopicPartitionToOffsetMetadataMap);
或将元数据指定为 Af//////////这意味着 NO_TIMESTAMP在 Kafka Streams 中。