获取kafka中topic偏移量和消费偏移量

 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改成和消费者一致时能成功查到。

注意:千万不要在另外一个服务以上面这种方式查询,要不然会创建一个消费者,但是没有消费代码。

相关推荐

  1. 获取kafkatopic偏移消费偏移

    2024-03-22 05:36:06       20 阅读
  2. Kafka消费者提交偏移

    2024-03-22 05:36:06       32 阅读
  3. 每期一个小窍门: 重置kafka 消费者偏移

    2024-03-22 05:36:06       33 阅读
  4. kafka 偏移的类型与提交方式

    2024-03-22 05:36:06       40 阅读

最近更新

  1. TCP协议是安全的吗?

    2024-03-22 05:36:06       18 阅读
  2. 阿里云服务器执行yum,一直下载docker-ce-stable失败

    2024-03-22 05:36:06       19 阅读
  3. 【Python教程】压缩PDF文件大小

    2024-03-22 05:36:06       19 阅读
  4. 通过文章id递归查询所有评论(xml)

    2024-03-22 05:36:06       20 阅读

热门阅读

  1. P1118 [USACO06FEB] Backward Digit Sums G/S

    2024-03-22 05:36:06       17 阅读
  2. ARM day6

    2024-03-22 05:36:06       19 阅读
  3. spring cloud gateway k8s优雅启停

    2024-03-22 05:36:06       18 阅读
  4. Flink:Lookup Join 实现与示例代码

    2024-03-22 05:36:06       18 阅读
  5. 深度学习Top10算法之ResNet

    2024-03-22 05:36:06       17 阅读
  6. flink自定义函数如何从崩溃中恢复数据

    2024-03-22 05:36:06       16 阅读
  7. FlinkSQL之保序任务对于聚合SQL影响分析

    2024-03-22 05:36:06       24 阅读