SpringBoot中如何通过Acknowledgement提交Kafka指定分区的Offset？

我正在使用ConcurrentMessageListenerContainer消费Kafka Topic的消息，单个消费者组内的消费者可被分配同一Topic的多个分区，配置代码如下：

@Configuration
@EnableKafka
public class KafkaConfig {

    @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
                        kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
                                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3);
        factory.getContainerProperties().setPollTimeout(3000);
        return factory;
    }

    @Bean
    public ConsumerFactory<Integer, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafka.getBrokersAsString());
        ...
        return props;
    }
}

我希望仅提交处理成功的特定分区的Offset，消费代码如下：

@KafkaListener(id = "thing2", topicPartitions =
        { @TopicPartition(topic = "topic1", partitions = { "0", "1" }),
          @TopicPartition(topic = "topic2", partitions = "0",
             partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "100"))
        })
public void listen(List<ConsumerRecord<String, String>> records, Acknowledgement acknowledgement) {
   Map<TopicPartition, List<ConsumerRecord>> recordsMap = ....
     
     acknowledgement.acknowledge();
    ...
}

但Acknowledgement没有提供指定分区提交Offset的选项，默认会提交所有分区的Offset。请问是否有办法实现仅提交指定分区Offset的需求？我查阅了Spring Kafka文档，发现其中没有指定分区提交Offset的相关方法。

要实现仅提交指定分区的Offset，可通过以下两种方式实现：

方式一：使用ConsumerAwareAcknowledgement替代Acknowledgement #

ConsumerAwareAcknowledgement是Acknowledgement的子类，它能获取底层Kafka Consumer实例，直接调用Consumer的commitSync(Map<TopicPartition, OffsetAndMetadata>)方法提交指定分区的Offset：

@KafkaListener(id = "thing2", topicPartitions =
        { @TopicPartition(topic = "topic1", partitions = { "0", "1" }),
          @TopicPartition(topic = "topic2", partitions = "0",
             partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "100"))
        })
public void listen(List<ConsumerRecord<String, String>> records, ConsumerAwareAcknowledgement ack) {
    // 按分区分组消息记录
    Map<TopicPartition, List<ConsumerRecord>> recordsMap = records.stream()
            .collect(Collectors.groupingBy(record -> 
                new TopicPartition(record.topic(), record.partition())));

    Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new HashMap<>();

    for (Map.Entry<TopicPartition, List<ConsumerRecord>> entry : recordsMap.entrySet()) {
        TopicPartition tp = entry.getKey();
        List<ConsumerRecord> partitionRecords = entry.getValue();
        
        // 处理当前分区消息，根据实际业务判断是否处理成功
        boolean processedSuccessfully = processPartitionRecords(partitionRecords);
        
        if (processedSuccessfully) {
            // 提交的Offset是当前分区最后一条消息的下一个位置
            long nextOffset = partitionRecords.get(partitionRecords.size() - 1).offset() + 1;
            offsetsToCommit.put(tp, new OffsetAndMetadata(nextOffset));
        }
    }

    // 仅提交处理成功的分区Offset
    if (!offsetsToCommit.isEmpty()) {
        ack.getConsumer().commitSync(offsetsToCommit);
    }
}

// 模拟分区消息处理逻辑
private boolean processPartitionRecords(List<ConsumerRecord> records) {
    // 你的业务处理代码
    return true;
}

方式二：自定义Offset管理（适合复杂场景） #

如果需要更灵活的Offset控制（比如将Offset存储到数据库），可按以下步骤操作：

1. 在消费者配置中关闭自动提交：props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
2. 不依赖容器的Acknowledgement，自行维护每个分区的成功Offset
3. 根据业务逻辑，在合适时机调用Consumer的commitSync/commitAsync提交指定分区，或实现自定义Offset存储与恢复逻辑

注意事项 #

• commitSync会阻塞直到提交完成，commitAsync为异步提交，需自行处理提交失败的重试逻辑
• 调用commit方法时，需确保当前Consumer仍持有对应分区的消费权限（未被容器重新分配）

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