如何通过Java Kafka Admin客户端获取Topic的retention.bytes与retention.ms

你找到的API方向是对的，以下是具体的参数传入方式和完整实现：

核心步骤说明 #

• 首先创建ConfigResource对象，指定资源类型为TOPIC，并传入目标Topic名称
• 调用describeConfigs方法，传入包含该ConfigResource的集合
• 从返回的配置结果中提取retention.bytes和retention.ms的值

完整代码示例 #

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.Config;
import org.apache.kafka.clients.admin.ConfigEntry;
import org.apache.kafka.clients.admin.ConfigResource;

import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class KafkaTopicConfigGetter {
    public static void main(String[] args) {
        // 配置AdminClient的连接参数
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "your-kafka-brokers:9092");
        
        try (AdminClient adminClient = AdminClient.create(props)) {
            // 创建目标Topic的ConfigResource
            String targetTopic = "your-topic-name";
            ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC, targetTopic);
            
            // 调用describeConfigs获取配置
            Config config = adminClient.describeConfigs(Collections.singletonList(resource))
                    .values()
                    .get(resource) // 传入之前创建的ConfigResource对象
                    .get(); // 阻塞获取异步结果
            
            // 提取retention配置
            ConfigEntry retentionBytesEntry = config.get("retention.bytes");
            ConfigEntry retentionMsEntry = config.get("retention.ms");
            
            // 输出配置值
            if (retentionBytesEntry != null) {
                System.out.println("retention.bytes: " + retentionBytesEntry.value());
            }
            if (retentionMsEntry != null) {
                System.out.println("retention.ms: " + retentionMsEntry.value());
            }
            
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }
}

参数说明 #

• ConfigResource构造参数：第一个是ConfigResource.Type.TOPIC（表示资源是Topic），第二个是你要查询的Topic名称
• describeConfigs的入参：需要传入包含目标ConfigResource的集合，示例中用Collections.singletonList(resource)生成单元素集合
• values().get(resource)：这里的resource就是你之前创建的那个ConfigResource对象，用来匹配查询结果
• 最后get("retention.bytes")和get("retention.ms")：就是你要获取的配置项名称，注意拼写是retention.bytes（复数），不是你猜测的retention.byte

内容的提问来源于stack exchange，提问作者paul