您对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）时，应在最后处理的消息的偏移量上添加一个。”
