如何在Java中使用Kafka Admin Client获取Kafka任意提交偏移量的提交时间？

在Java中使用Kafka Admin Client获取Kafka任意提交偏移量的提交时间，可以按照以下步骤进行：

首先，确保已经引入了Kafka的Java客户端依赖库。可以在项目的pom.xml文件中添加以下依赖：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>${kafka.version}</version>
</dependency>

其中${kafka.version}是Kafka客户端的版本号。

创建Kafka Admin Client实例，并配置所需的Kafka集群连接信息。示例代码如下：

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.DescribeConsumerGroupOptions;
import org.apache.kafka.clients.admin.DescribeConsumerGroupsResult;
import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsResult;
import org.apache.kafka.clients.admin.ListConsumerGroupsOptions;
import org.apache.kafka.clients.admin.ListConsumerGroupsResult;
import org.apache.kafka.clients.admin.OffsetSpec;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.consumer.OffsetCommitCallback;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ExecutionException;

public class KafkaOffsetTimeFetcher {

    private static final String BOOTSTRAP_SERVERS = "kafka-bootstrap-server:9092";
    private static final String GROUP_ID = "your-consumer-group-id";
    private static final String TOPIC_NAME = "your-topic-name";
    private static final int PARTITION = 0;
    private static final Duration TIMEOUT = Duration.ofSeconds(10);

    public static void main(String[] args) {
        // 创建Kafka Admin Client配置
        Properties properties = new Properties();
        properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        properties.put(AdminClientConfig.CLIENT_ID_CONFIG, "kafka-offset-fetcher");
        
        // 创建Kafka Admin Client实例
        try (AdminClient adminClient = AdminClient.create(properties)) {
            // 获取Consumer Group的信息
            DescribeConsumerGroupOptions groupOptions = new DescribeConsumerGroupOptions()
                    .timeoutMs((int) TIMEOUT.toMillis());
            KafkaFuture<Set<String>> groupIdsFuture = adminClient.listConsumerGroups(groupOptions).all();
            Set<String> groupIds = groupIdsFuture.get();
            System.out.println("Consumer Groups: " + groupIds);
            
            // 获取指定Consumer Group的提交偏移量信息
            if (groupIds.contains(GROUP_ID)) {
                ListConsumerGroupsOptions listOptions = new ListConsumerGroupsOptions()
                        .timeoutMs((int) TIMEOUT.toMillis());
                KafkaFuture<ListConsumerGroupOffsetsResult> offsetsFuture = adminClient
                        .listConsumerGroupOffsets(GROUP_ID, listOptions).partitionsToOffsetAndMetadata()
                        .get(new TopicPartition(TOPIC_NAME, PARTITION));
                OffsetAndMetadata offsetAndMetadata = offsetsFuture.get().offsets().get(
                        new TopicPartition(TOPIC_NAME, PARTITION));
                long offset = offsetAndMetadata.offset();
                long commitTimestamp = offsetAndMetadata.commitTimestamp();
                System.out.println("Offset: " + offset);
                System.out.println("Commit Timestamp: " + commitTimestamp);
            } else {
                System.out.println("Consumer Group " + GROUP_ID + " not found.");
            }
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }
}

注意替换代码中的BOOTSTRAP_SERVERS、GROUP_ID和TOPIC_NAME为相应的Kafka集群连接地址、消费者组ID和主题名称。