Kafka流无法解码StreamTask中的时间戳元数据

Question

在启动应用程序时，Kafka流出现了奇怪的错误

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)

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

根据org内部的代码。阿帕奇。Kafka。溪流。加工机内部。StreamTask，由于解码时间戳元数据时出错而发生故障（StreamTask.decodeTimestamp（））。它发生在prod上，不能在舞台上复制。这些错误的根本原因是什么？

额外信息：我们的应用程序使用Kafka流，并使用相同的应用程序使用来自多个Kafka代理的消息。id和状态。dir（实际上，我们从一个代理切换到另一个代理，但在一段时间内，我们连接到两个代理，所以我们有两个Kafka流，每个代理一个）。据我所知，消费者群体生活在代理端（所以不应该是问题），但state dir生活在客户端。可能由于使用了相同的状态而出现了一些竞争条件。两条Kafka流的dir？这可能是根本原因吗？

我们使用kafka-streamv.2.4.0、kafka-客户端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": "", "host name"："uniting-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)));

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

>

将消费者组更改为新用户组。请注意，根据最新或最早偏移重置策略，您可能会丢失或重复消息。因此，对于某些情况，此选项可能是不可接受的

手动覆盖元数据（时间戳根据StreamTask.decodeTimestamp（）中的逻辑编码）：

地图