在一个spring项目中,我使用了kafka,现在我想创建一个以“topicname”和“groupeid”为参数的方法,并计算“topic分区的最后偏移量”和“组消耗的偏移量”之间的差异
对于最后的偏移量,我现在需要得到消耗的偏移量来计算差值
public ResponseEntity<Offsets> deltaoffsets (@RequestParam( name = "groupId") String groupId, @RequestParam( name = "topic") String topic) {
Map<String,Object> properties = (Map) kafkaLocalConsumerConfig.get("kafkaLocalConsumerConfig");
properties.put("group.id", groupId);
properties.put("enable.auto.commit", "true");
List<TopicPartition> partition=new ArrayList<>();
KafkaConsumer<String, RefentialToReload> kafkaLocalConsumer = new KafkaConsumer<>(properties);
Map<String, List<PartitionInfo>> topics = kafkaLocalConsumer.listTopics();
List<PartitionInfo> partitionInfos = topics.get(topic);
if (partitionInfos == null) {
log.warn("Partition information was not found for topic");
}
else {
for (PartitionInfo partitionInfo : partitionInfos) {
TopicPartition topicPartition = new TopicPartition(topic, partitionInfo.partition());
partition.add(topicPartition);
log.info("partition assigned to kafkaLocalConsumer");
}
}
//get lastOffsets of the topicPartition
Map<TopicPartition,Long> OffsetsTopicpartition = kafkaLocalConsumer.endOffsets(kafkaLocalConsumer.assignment());
//here i need to get consumed offsets
}
1条答案
按热度按时间drnojrws1#
beginningOffsets()
是第一个,不是最后一个。你可以使用
AdminClient
-下面是一个显示当前偏移和结束偏移的示例。。。