Apache Kafka压缩主题：定期重置消费者偏移量及获取同键新值

首先纠正一个认知：压缩主题中同键的新消息会拥有全新的偏移量，旧消息会被标记为待清理，在后台压缩过程中移除，但新消息是独立的新条目。消费者如果已经提交过旧键消息的偏移量，默认不会自动重新读取这些键的新消息，以下是几种无需更换group-id即可获取新值的方案：

方案1：手动重置指定分区的偏移量到起始位置 #

如果需要定期重新读取主题的全部内容（包括压缩后的最新同键值），可以通过Spring Kafka的API手动重置偏移量：

@Autowired
private KafkaListenerEndpointRegistry registry;

public void resetOffsetsToStart() {
    // 获取目标消费者容器
    MessageListenerContainer container = registry.getListenerContainer("your-listener-id");
    if (container != null && container.isRunning()) {
        container.stop();
        // 注册重平衡监听器，在分区分配后重置偏移量到起始位置
        container.getContainerProperties().setConsumerRebalanceListener(new ConsumerRebalanceListener() {
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {}

            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                container.getConsumer().seekToBeginning(partitions);
            }
        });
        container.start();
    }
}

• 替换your-listener-id为你在@KafkaListener注解中指定的id属性值。
• 该操作会重启消费者容器，让其重新读取主题中所有存活的消息（即压缩后保留的每个键的最新值）。

方案2：基于时间戳重置偏移量（按需读取某段时间后的新消息） #

如果不需要从头读取，只想获取某个时间点之后的同键更新，可以通过时间戳计算对应偏移量：

@Autowired
private KafkaListenerEndpointRegistry registry;

public void resetOffsetsByTimestamp(String topic, long timestamp) {
    MessageListenerContainer container = registry.getListenerContainer("your-listener-id");
    if (container != null && container.isRunning()) {
        container.stop();
        container.getContainerProperties().setConsumerRebalanceListener(new ConsumerRebalanceListener() {
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {}

            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                KafkaConsumer<?, ?> consumer = container.getConsumer();
                Map<TopicPartition, Long> timestampMap = new HashMap<>();
                partitions.forEach(tp -> timestampMap.put(tp, timestamp));
                
                // 根据时间戳查询对应的偏移量
                Map<TopicPartition, OffsetAndTimestamp> offsetMap = consumer.offsetsForTimes(timestampMap);
                for (TopicPartition tp : partitions) {
                    OffsetAndTimestamp offsetInfo = offsetMap.get(tp);
                    if (offsetInfo != null) {
                        consumer.seek(tp, offsetInfo.offset());
                    } else {
                        // 无对应时间戳的消息时，跳到分区末尾
                        consumer.seekToEnd(Collections.singleton(tp));
                    }
                }
            }
        });
        container.start();
    }
}

• 传入的timestamp为毫秒级时间戳，消费者会从该时间点之后的消息开始消费。

方案3：结合定时任务按需启动并重置偏移量 #

将消费者容器设置为非自动启动，通过定时任务触发重置和消费：

@KafkaListener(id = "scheduled-compacted-listener", topics = "your-compacted-topic", autoStartup = "false")
public void consumeCompactedTopic(ConsumerRecord<String, YourEntity> record) {
    // 消息处理逻辑
}

@Scheduled(fixedRate = 3600000) // 每小时执行一次
public void scheduledResetAndConsume() {
    resetOffsetsToStart(); // 调用方案1的重置方法
    registry.getListenerContainer("scheduled-compacted-listener").start();
    // 可根据需求在消费完成后停止容器
}

• 需在Spring Boot启动类上添加@EnableScheduling注解启用定时任务。

关键注意事项 #

• 压缩主题的核心是保留每个键的最新值，若仅需获取当前所有键的最新状态，一次性从头消费是高效的选择——压缩后的主题只保留每个键的最新消息，数据量不会过大。
• 重置偏移量前必须停止消费者容器，避免并发冲突。

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