Kafkajava

yrefmtwq 于 2021-06-27 发布在 Java

关注(0)|答案(1)|浏览(211)

在一个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 
}

Java apache-kafka spring-boot spring-kafka

来源：https://stackoverflow.com/questions/65566470/calculate-delta-offsets-kafka-java

1条答案

按热度按时间

drnojrws1#

beginningOffsets() 是第一个，不是最后一个。
你可以使用 AdminClient -下面是一个显示当前偏移和结束偏移的示例。。。

@Bean
public ApplicationRunner runner(KafkaAdmin admin, ConsumerFactory<String, String> cf) throws Exception {
    return args -> {
    try (
            AdminClient client = AdminClient.create(admin.getConfig());
            Consumer<String, String> consumer = cf.createConsumer("group", "clientId", "");
        ) {
        Collection<ConsumerGroupListing> groups = client.listConsumerGroups()
                .all()
                .get(10, TimeUnit.SECONDS);
        groups.forEach(group -> {
            Map<TopicPartition, OffsetAndMetadata> map = null;
            try {
                map = client.listConsumerGroupOffsets(group.groupId())
                        .partitionsToOffsetAndMetadata()
                        .get(10, TimeUnit.SECONDS);
            }
            catch (InterruptedException e) {
                e.printStackTrace();
                Thread.currentThread().interrupt();
            }
            catch (ExecutionException e) {
                e.printStackTrace();
            }
            catch (TimeoutException e) {
                e.printStackTrace();
            }
            Map<TopicPartition, Long> endOffsets = consumer.endOffsets(map.keySet());
            map.forEach((tp, off) -> {
                System.out.println("group: " + group + " tp: " + tp
                        + " current offset: " + off.offset()
                        + " end offset: " + endOffsets.get(tp));
            });
        });
    }
    };
}

赞(0）回复(0）举报 2021-06-27

我来回答

Kafkajava

1条答案

相关问题

热门标签

最新问答