如何在Kafka Admin API中添加主题专属配置？详解configs传参

没问题，这其实很简单，你只需要先构建一个包含主题配置键值对的Map，然后通过NewTopic的configs()方法把这个Map绑定到主题上就行。我给你整理了完整的代码示例，一步步来：

步骤1：构建主题配置Map #

先创建一个HashMap（或者其他Map实现类），把你需要的主题配置以键值对的形式放进去，比如常见的清理策略、消息保留时长等：

import java.util.HashMap;
import java.util.Map;

// 构建主题专属配置
Map<String, String> topicConfigs = new HashMap<>();
// 示例1：设置消息清理策略为"删除"（默认就是delete，这里只是演示）
topicConfigs.put("cleanup.policy", "delete");
// 示例2：设置消息保留时长为7天（单位毫秒）
topicConfigs.put("retention.ms", "604800000");
// 你可以根据需求添加更多配置，比如segment.bytes、min.insync.replicas等

步骤2：创建带配置的NewTopic对象 #

你可以用链式调用的方式，在实例化NewTopic后直接调用configs()方法传入配置Map，这样代码更简洁：

import org.apache.kafka.clients.admin.NewTopic;
import java.util.ArrayList;
import java.util.List;

// 创建带配置的NewTopic实例
NewTopic newTopic = new NewTopic("topicName", getPartitionCount(), getReplicationFactor())
        .configs(topicConfigs); // 注入配置Map

// 将主题加入创建列表
List<NewTopic> newKafkaTopicsList = new ArrayList<>();
newKafkaTopicsList.add(newTopic);

如果需要后续修改配置，也可以先创建NewTopic对象，再单独调用configs()方法：

NewTopic newTopic = new NewTopic("topicName", getPartitionCount(), getReplicationFactor());
newTopic.configs(topicConfigs); // 单独设置配置

步骤3：调用AdminClient创建主题 #

最后还是用你原来的代码调用createTopics接口即可：

CreateTopicsResult createTopicsResult = adminClient.createTopics(newKafkaTopicsList);

需要注意的是，configs()方法会返回NewTopic本身，所以链式调用是完全可行的，这样写出来的代码更紧凑易读。你可以根据业务需求添加任意Kafka支持的主题级配置。

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