Apache Kafka Client(Java):列出主题并检查主题是否经过日志压缩 [英] Apache Kafka Client (Java) : List topics and check whether topic is log compacted

问题描述

背景

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

We have Apache Kafka managed by Zookeeper in our company. One of our Spring Boot applications needs to check the list of all topic available and also list which ones have log compaction enabled (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命令时得到的响应相同的响应.

With the above code the application can get the list of topics. Is there a way to also get to know whether the individual topics are log compacted ? What I am looking for is some "Java" way to get the same response that I get when I run the following Apache Kafka CLI command from terminal.

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

推荐答案

您应使用

You should use the AdminClient API to retrieve that information.

• 首先使用 listTopics() 检索主题列表.
• 然后使用

• First use listTopics() to retrieve the topic list.
• Then use describeConfigs() to get the configuration for each topic.
• Finally, out of the ConfigEntry objects you'll get, you can then filter the topics that have compact as the cleanup.policy.

这基本上是 kafka-topics 工具的工作，因此您可以查看其源代码

This is basically what the kafka-topics tools is doing, so you can have a look at its source kafka.admin.TopicCommand. Even though it's Scala, the concepts are similar.

这篇关于Apache Kafka Client(Java):列出主题并检查主题是否经过日志压缩的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！