如何使用Kafka Streams统计topic消息并限量迁移避免解码无限循环

核心解决思路 #

要实现「只迁移A-errors中调用接口那一刻已存在的所有消息，迁移完成立即关闭流，避免后续新进入A-errors的消息被重复处理」的需求，有两种可行的实现方案，推荐第一种更轻量的方案：

方案1：基于Kafka消费者获取当前分区结束偏移量，消费到偏移量后关闭流 #

这个方案不需要修改现有流逻辑，只需要增加偏移量检查的逻辑，步骤如下：

1. 调用接口时先获取A-errors所有分区的当前结束偏移量
   用Kafka AdminClient或者原生Consumer先查询A-errors每个分区的最大偏移量，这就是本次需要迁移的总消息数的基准，只有偏移量小于这个值的消息才需要迁移。
2. 在流处理中增加计数和关闭逻辑
   在foreach算子中统计已经处理的消息数，当所有分区都消费到之前查询到的结束偏移量时，主动调用streams.close()关闭流。
3. 避免无限循环的配置

• 每次调用接口时，生成唯一的Application ID，不要复用固定的"test"，这样每次调用都会从A-errors的最早偏移量开始消费，不会受之前的消费记录影响
• 解码逻辑中如果再次失败，将消息发回A-errors时不需要做特殊处理，因为本次流已经在迁移完旧消息后关闭了，新写入A-errors的消息不会被本次流消费，直到下次调用接口才会被处理。

方案2：直接用Kafka原生消费者+生产者实现，比Kafka Streams更适合这种一次性迁移场景 #

如果不需要Kafka Streams的其他能力，一次性的消息迁移用原生客户端更简单，不需要处理流的生命周期问题：

1. 消费者订阅A-errors，配置auto.offset.reset=earliest，关闭自动提交
2. 批量拉取消息，生产到A topic，生产成功后提交消费偏移量
3. 当拉取的消息为空，且消费偏移量已经达到调用接口时查询到的分区最大偏移量时，关闭消费者和生产者即可。

现有代码修改示例（基于方案1） #

public void initializeKafkaStreams() {
    String inputTopic = "A-errors";
    String outputTopic = "A";
    // 每次生成唯一的application id，避免复用之前的消费偏移量
    String uniqueAppId = "error-replay-" + UUID.randomUUID();
    Properties streamsConfiguration = new Properties();
    streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, uniqueAppId);
    streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
    streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
    // 配置从最早偏移量开始消费
    streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    try {
        Path stateDirectory = Files.createTempDirectory("kafka-streams");
        streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, stateDirectory.toAbsolutePath().toString());
    } catch (IOException e) {
        e.printStackTrace();
    }

    // 第一步：先查询A-errors所有分区的当前最大偏移量，作为本次迁移的结束标记
    Map<TopicPartition, Long> endOffsets;
    try (AdminClient admin = AdminClient.create(streamsConfiguration)) {
        List<PartitionInfo> partitions = admin.describeTopics(Collections.singleton(inputTopic))
                .allTopicNames().get().get(inputTopic).partitions();
        List<TopicPartition> topicPartitions = partitions.stream()
                .map(p -> new TopicPartition(inputTopic, p.partition()))
                .collect(Collectors.toList());
        endOffsets = admin.endOffsets(topicPartitions).all().get();
    } catch (Exception e) {
        e.printStackTrace();
        return;
    }
    // 记录已处理的每个分区的偏移量
    Map<TopicPartition, Long> processedOffsets = new ConcurrentHashMap<>();
    AtomicBoolean allProcessed = new AtomicBoolean(false);

    //Building a Streaming Topology
    StreamsBuilder builder = new StreamsBuilder();
    KStream<String, String> rawMessages = builder.stream(inputTopic);

    rawMessages.foreach((devId, obj, recordContext) -> {
        System.out.println(obj);
        TopicPartition tp = new TopicPartition(inputTopic, recordContext.partition());
        processedOffsets.put(tp, recordContext.offset());
        // 检查是否所有分区都已经消费到结束偏移量
        boolean finished = endOffsets.entrySet().stream()
                .allMatch(e -> processedOffsets.getOrDefault(e.getKey(), -1L) >= e.getValue() - 1);
        if (finished) {
            allProcessed.set(true);
        }
    });
    //Handling Results
    rawMessages.to(outputTopic, Produced.with(Serdes.String(), Serdes.String()));
    //Starting KafkaStream Job
    Topology topology = builder.build();
    KafkaStreams streams = new KafkaStreams(topology, streamsConfiguration);
    streams.start();

    // 异步检查是否处理完成，完成后关闭流
    new Thread(() -> {
        while (!allProcessed.get()) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }
        streams.close(Duration.ofSeconds(10));
        // 可选：删除临时状态目录，清理资源
    }).start();
}

额外注意点 #

• 生产环境建议增加超时逻辑，如果长时间没有消费到结束偏移量，主动关闭流避免资源泄漏
• 如果要保证消息不丢失不重复，可以开启Kafka Streams的事务配置，保证生产到A和消费偏移量提交的原子性
• 解码失败后发回A-errors的消息，会作为新消息存在，下次调用接口时才会被迁移，不会出现无限循环的问题

内容的提问来源于stack exchange，提问作者BrentL