1、kafkaclient版本1.0.1
public class MutiThreadScheduleTask {
@Resource
private KafkaConsumer<String, String> kafkaConsumer;
public void test(String topic) {
//查询 topic partitions
List<TopicPartition> topicPartitionList = new ArrayList<>();
List<PartitionInfo> partitionInfoList = kafkaConsumer.partitionsFor(topic);
for (PartitionInfo partitionInfo : partitionInfoList) {
TopicPartition topicPartition = new TopicPartition(partitionInfo.topic(), partitionInfo.partition());
topicPartitionList.add(topicPartition);
}
//查询总偏移量
Map<Integer, Long> endOffsetMap = new HashMap<>();
Map<TopicPartition, Long> endOffsets = kafkaConsumer.endOffsets(topicPartitionList);
for (TopicPartition partitionInfo : endOffsets.keySet()) {
endOffsetMap.put(partitionInfo.partition(), endOffsets.get(partitionInfo));
}
//查询消费消费偏移量
Map<Integer, Long> commitOffsetMap = new HashMap<>();
for (TopicPartition topicAndPartition : topicPartitionList) {
OffsetAndMetadata committed = kafkaConsumer.committed(topicAndPartition);
commitOffsetMap.put(topicAndPartition.partition(), committed.offset());
}
}
}
2、这部分测试代码和真正的消费端不在一个程序内,连接kafka时随便配置了一个groupId,然后kafkaConsumer.committed(topicAndPartition)一直为null。后来groupId改成和消费者一致时能成功查到。
注意:千万不要在另外一个服务以上面这种方式查询,要不然会创建一个消费者,触发Kafka的再均衡。