Apache Kafka客户端（Java）：列出主题并检查主题是否被日志压缩

Question

出身背景

我们公司有由动物园管理员管理的阿帕奇·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

首先需要创建一个带有属性文件的kafka管理客户端。

import org.apache.kafka.clients.admin.AdminClient;
import java.util.*;
import org.apache.kafka.clients.admin.Config;
import org.apache.kafka.clients.admin.ConfigEntry;
import org.apache.kafka.clients.admin.DescribeConfigsResult;
import org.apache.kafka.common.config.ConfigResource;

admin = AdminClient.create(properties);

然后需要使用管理员。describeConfigs方法，此方法将ConfigResources的集合作为参数，定义如下。

Collection<ConfigResource> cr =  Collections.singleton( new ConfigResource(ConfigResource.Type.TOPIC, "<Your Topic Name>") 

DescribeConfigsResult ConfigsResult = admin.describeConfigs(cr));

使用ConfigsResult，您需要评估和解析此数据类型，在这里我将其硬转换为配置数据类型。

Config all_configs = (Config)ConfigsResult.all().get().values().toArray()[0];

然后需要遍历包含许多ConfigResource对象的all\u configs数据类型。下面的代码构建了一个迭代器，遍历并查找用于清理的配置策略的配置值，您可以使用相同的逻辑来查找其他字段的配置值。

    Iterator ConfigIterator = all_configs.entries().iterator();

    while (ConfigIterator.hasNext()) 
    {
          ConfigEntry currentConfig = (ConfigEntry) ConfigIterator.next();
          if (currentConfig.name().equals("cleanup.policy")) {
                System.out.println(currentConfig.value());
          }
        }

Answer 2

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

• 首先使用listTopics（）检索主题列表
• 然后使用describeConfigs（）获取每个主题的配置
• 最后，从您将获得的ConfigEntry对象中，您可以过滤具有压缩的主题作为清理。政策

这基本上就是Kafka主题工具所做的，因此您可以查看其源代码Kafka。管理TopicCommand。尽管它是Scala，但概念是相似的。