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
            OffsetAndMetadata offset = consumer.committed(tp);
            consumer.seek(tp, offset.offset() - 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();
        }
 }