java - Apache Kafka 客户端(Java): List topics and check whether topic is log compacted

Question

背景

我们公司有由 Zookeeper 管理的 Apache Kafka。我们的 Spring Boot 应用程序之一需要检查所有可用主题的列表，并列出哪些主题启用了日志压缩 (cleanup.policy=compact)。

当前代码

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBrokerList);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaConsumerGroup);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        return new DefaultKafkaConsumerFactory<>(props);
    }

...
...

    public List<String> getTopics() {
        Map<String, List<PartitionInfo>> topics = consumerFactory().createConsumer().listTopics();
        List<String> topicList = new ArrayList<>();
        topics.keySet().remove(CONSUMER_OFFSETS);
        topicList.addAll(topics.keySet());
        return topicList;
    }

问题

通过上面的代码，应用程序可以获得主题列表。有没有办法也了解各个主题是否经过日志压缩？我正在寻找某种“Java”方式来获得与从终端运行以下 Apache Kafka CLI 命令时得到的相同响应。

kafka-topics --zookeeper localhost:2181 --describe --topic TestTopicCompact

对此的响应示例

Topic:TestTopicCompact  PartitionCount:1    ReplicationFactor:1 Configs:cleanup.policy=compact
    Topic: TestTopicCompact Partition: 0    Leader: 1001    Replicas: 1001  Isr: 1001

Answer 1

您应该使用AdminClient API检索该信息。

• 首次使用listTopics()检索主题列表。
• 然后使用 describeConfigs()获取每个主题的配置。
• 最后，从您将获得的 ConfigEntry 对象中，您可以过滤将 compact 作为 cleanup.policy 的主题。

这基本上就是 kafka-topics 工具正在做的事情，因此您可以查看其源代码 kafka.admin.TopicCommand 。尽管它是 Scala，但概念是相似的。