zoukankan      html  css  js  c++  java
  • Java API获取consumer group最新提交位移的时间

    碰到了有人问起这个问题,目前java consumer没有利用OffsetAndMetadata中的metadata字段记录提交的时间,故直接通过java consumer来查询是不行,我们需要直接读取__consumer_offsets,代码如下:

    public static long lastCommitTimestamp(String groupName, String bootstrapServers) {
        int partition = Math.abs(groupName.hashCode() % 50); // Assume offset topic partition number is 50
        TopicPartition tp = new TopicPartition("__consumer_offsets", partition);
    
        Properties props = new Properties();
        props.put("bootstrap.servers", bootstrapServers);
        props.put("group.id", "test");
        props.put("enable.auto.commit", "false");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
        	consumer.assign(Arrays.asList(tp));
        	consumer.poll(Duration.ZERO); // take effect
    
        	Map<TopicPartition, Long> endOffset = consumer.endOffsets(Collections.singletonList(tp));
        	consumer.seek(tp, endOffset.get(tp) - 1);
        	
        	ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(30));
        	if (records.count() == 0) { 
        		throw new IllegalStateException("Consumer group does not commit offsets."); 
        	} 
        	return records.iterator().next().timestamp(); 
        } 
    }
    

      

  • 相关阅读:
    adb devices检测不到夜神模拟器
    adb devices检测不到夜神模拟器
    adb devices检测不到夜神模拟器
    epoll里面mmap释疑
    epoll里面mmap释疑
    epoll里面mmap释疑
    epoll里面mmap释疑
    Redis数据迁移的三个方法
    Redis数据迁移的三个方法
    MySQL:由USE DB堵塞故障引发的思考
  • 原文地址:https://www.cnblogs.com/huxi2b/p/10213428.html
Copyright © 2011-2022 走看看