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(); 
    } 
}

  


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM