如何获取每个分区的最新偏移量，然后仅消耗该偏移量？

您可以使用consumer.partitionsFor和consumer.endOffsets来获取分区和上一个偏移量

分区

/*Get metadata about the partitions for a given topic. This method will issue a remote call to the server if it does  not already have any metadata about the given topic.*/ 
    public java.util.List<PartitionInfo> partitionsFor(java.lang.String topic)

endOffsets

/*Get the last offset for the given partitions. The last offset of a partition is the offset of the upcoming message, i.e. the offset of the last available message + 1*/
public java.util.Map<TopicPartition,java.lang.Long> endOffsets(java.util.Collection<TopicPartition> partitions)

。

以下是示例代码

Properties consumerProperties = new Properties();
consumerProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
consumerProperties.put(ConsumerConfig.GROUP_ID_CONFIG, "consumerid");
Consumer<String, byte[]> consumer = new KafkaConsumer<>(consumerProperties);    
List<PartitionInfo> parts = consumer.partitionsFor(topic);
consumer.assign(partitions);
Map<TopicPartition, Long> offsets = consumer.endOffsets(partitions);
for (TopicPartition tp : offsets.keySet()) {
    OffsetAndMetadata commitOffset = consumer.committed(new 
    TopicPartition(tp.topic(), tp.partition()));
    //Consumer offset for partition tp
    long offset=offsets.get(tp);
    //Consumed committed offset
    long consumedOffset=commitOffset.offset();
}