您对commitSync（）的调用将确认最后一个poll（）中批处理中的所有消息，而不是在处理消息时确认每个消息，这是我认为您要尝试的。
从文档中
“以上示例使用commitSync将所有收到的记录标记为已提交。在某些情况下，您可能希望通过显式指定偏移量来更好地控制已提交的记录。在下面的示例中，我们在处理完每个分区中的记录后提交偏移量。
try {
while(running) {
ConsumerRecords<String, String> records = consumer.poll(Long.MAX_VALUE);
for (TopicPartition partition : records.partitions()) {
List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
for (ConsumerRecord<String, String> record : partitionRecords) {
System.out.println(record.offset() + ": " + record.value());
}
long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(lastOffset + 1)));
}
}
} finally {
consumer.close();
}
注意：提交的偏移量应始终是应用程序将读取的下一条消息的偏移量。因此，在调用commitSync（offsets）时，应在最后处理的消息的偏移量上添加一个。”
602392714
清零编程群