Kafka 之 AdminClient API

目录

一. 前言

二. KafkaAdminClient API

2.1. API 总览

2.2. Topic 操作

2.2.1. 创建 Topic

2.2.2. Topic 列表

2.2.3. 删除 Topic

2.2.4. 描述 Topic 详细信息

2.3. 分区 Partition 操作

2.3.1. 增加分区

2.3.2. 分区副本重新分配

2.3.3. 查询分区副本列表

2.4. Config 配置信息

2.4.1. 描述配置详细信息

2.4.2. 修改配置信息

2.4.3. incrementalAlterConfigs

2.5. Scram 账户操作

2.5.1. 创建 Scram 账户

2.5.2. 删除 Scram 账户

2.5.3. 查询 Scram 账户信息

2.6. Acl 操作

2.6.1. 创建 Acl

2.6.2. 删除 Acl

---

一. 前言

    自0.11.0.0版本起，Kafka 社区推出了 AdminClient 和 KafkaAdminClient，意在统一所有的集群管理 API。使用 0.11.0.0 及以后版本的用户应该始终使用这个类来管理集群。虽然和原先服务器端的 AdminClient 类同名，但这个工具是属于客户端的，因此只需要在管理程序项目中添加 kafka-clients 依赖即可：

<dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>3.6.1</version>
</dependency>

    Kafka 的管理 Java 客户端，支持管理和检查 Topic、Broker、配置和 ACL。所需的最小 Broker版本是 0.10.0.0。有更严格要求的方法将指定所需的最小 Broker 版本。这个客户端是在 0.11.0.0中引入的，API 还在不断发展。我们将尝试以兼容的方式演进 API，但我们保留在必要时在次要版本中进行破坏性更改的权利。一旦 API 被认为是稳定的，我们将更新 InterfaceStability 注解和本通知。 

二. KafkaAdminClient API

2.1. API 总览

方法名称	方法说明
createTopics	创建一批新主题。此操作不是事务性的，因此它对于某些主题可能成功，而对于其他主题则可能失败。此方法返回成功后，可能需要几秒钟的时间，所有代理才能知道主题已创建。在这段时间内，listTopics（）和describeTopics（Collection）可能不会返回有关新主题的信息。版本0.10.1.0或更高版本的代理支持此操作。版本0.10.2.0支持validateOnly选项。
deleteTopics	删除一批主题。此操作不是事务性的，因此它对于某些主题可能成功，而对于其他主题则可能失败。AdminClient＃deleteTopics返回成功后，可能需要几秒钟，所有代理才能意识到主题已消失。在此期间，AdminClient＃listTopics和AdminClient＃describeTopics可能会继续返回有关已删除主题的信息。如果代理上的delete.topic.enable为false，则deleteTopics将标记主题为删除，但实际上不会删除它们。在这种情况下，期货将成功返回。版本0.10.1.0或更高版本的代理支持此操作。
listTopics	查询Topic列表
describeTopics	描述Topic的详情信息
describeCluster	描述Cluster的详情信息
describeAcls	根据提供的过滤器列出访问控制列表（ACL）。注意：createAcls或deleteAcls所做的更改可能需要一些时间才能反映到describeAcls的输出中。版本0.11.0.0或更高版本的代理支持此操作。
createAcls	创建绑定到特定资源的访问控制列表（ACL）。此操作不是事务性的，因此对于某些ACL可能成功，而对于其他ACL则可能失败。如果您尝试添加与现有ACL复制的ACL，则不会引发任何错误，但不会进行任何更改。版本0.11.0.0或更高版本的代理支持此操作。
deleteAcls	根据提供的过滤器删除访问控制列表（ACL）。此操作不是事务性的，因此对于某些ACL可能成功，而对于其他ACL则可能失败。版本0.11.0.0或更高版本的代理支持此操作。
describeConfigs	获取指定资源的配置。返回的配置包括默认值，isDefault（）方法可用于将其与用户提供的值区分开。isSensitive（）为true的配置条目的值始终为null，因此不会公开敏感信息。isReadOnly（）为true的配置条目无法更新。版本0.11.0.0或更高版本的代理支持此操作。
alterConfigs	使用默认选项更新指定资源的配置。更新不是事务性的，因此更新可能会为某些资源成功而为其他资源失败。特定资源的配置会自动更新。版本0.11.0.0或更高版本的代理支持此操作。
incrementalAlterConfigs	新版本中是使用incrementalAlterConfigs方法来修改Topic的配置项，该方法使用起来相对于alterConfigs要略微复杂一些，但因此功能更多、更灵活。
alterReplicaLogDirs	更改指定副本的日志目录。当前仅当在代理上创建副本之前使用此API时，此API才有用。它将支持在完全实施KIP-113之后已经创建的移动副本。此操作不是事务性的，因此它可能对某些副本成功而对其他副本失败。版本1.0.0或更高版本的代理支持此操作。
describeLogDirs	查询给定代理集上所有日志目录的信息版本1.0.0或更高版本的代理支持此操作。
describeReplicaLogDirs	查询副本日志目录信息中的指定副本。版本1.0.0或更高版本的代理支持此操作。
createPartitions	根据相应的值，增加作为newPartitions关键字指定的主题的分区数。如果为具有关键字的主题增加了分区，则分区逻辑或消息的顺序将受到影响。
deleteRecords	删除消息
createDelegationToken	创建DelegationToken
renewDelegationToken	更新DelegationToken
expireDelegationToken	过期DelegationToken
describeDelegationToken	描述DelegationToken的详情信息
describeConsumerGroups	描述消费组的详情信息
listConsumerGroups	查询消费组列表
listConsumerGroupOffsets	查询消费组的offset列表
deleteConsumerGroups	删除消费组，只能是闲置的消费组
deleteConsumerGroupOffsets	删除消费组中某些分区的已提交偏移量。只有当组没有主动订阅相应的Topic时，才会成功。
electLeaders	选举Leader
alterPartitionReassignments	修改分区副本
listPartitionReassignments	查询分区副本列表
removeMembersFromConsumerGroup	从消费组中移除消费者成员
alterConsumerGroupOffsets	修改消费组的offset
listOffsets	查询Topic分区的offset列表
describeClientQuotas	描述客户端配额配置详细信息
alterClientQuotas	修改客户端配额
describeUserScramCredentials	描述Scram账户详细信息
alterUserScramCredentials	修改Scram账户
describeFeatures	描述已完成的功能以及支持的功能
updateFeatures	将指定的更新应用于最终确定的功能
describeMetadataQuorum	描述元数据数量的状态
unregisterBroker	注销Broker
describeProducers	描述生产者详细信息
describeTransactions	描述事务的详细信息
abortTransaction	终止事务
listTransactions	查询事务列表
fenceProducers	隔离使用任何提供的事务ID的所有活动生产者
metrics	获取adminClient保存的度量值

2.2. Topic 操作

2.2.1. 创建 Topic

// bootstrapServers 如 localhost:9092
private void createTopics(String bootstrapServers) {Properties properties = new Properties();properties.put("bootstrap.servers", bootstrapServers);properties.put("connections.max.idle.ms", 10000);properties.put("request.timeout.ms", 5000);try (AdminClient client = AdminClient.create(properties)) {CreateTopicsResult result = client.createTopics(Arrays.asList(new NewTopic("topic1", 1, (short) 1),new NewTopic("topic2", 1, (short) 1),new NewTopic("topic3", 1, (short) 1)));try {result.all().get();} catch (InterruptedException | ExecutionException e) {throw new IllegalStateException(e);}}
}

2.2.2. Topic 列表

private void listTopics(String bootstrapServers) {Properties properties = new Properties();properties.put("bootstrap.servers", bootstrapServers);properties.put("connections.max.idle.ms", 10000);properties.put("request.timeout.ms", 5000);try (AdminClient client = AdminClient.create(properties)) {ListTopicsResult result = client.listTopics();try {result.listings().get().forEach(topic -> {System.out.println(topic);});} catch (InterruptedException | ExecutionException e) {throw new IllegalStateException(e);}}
}// 运行结果
(name=topic1, internal=false)
(name=topic2, internal=false)
(name=topic3, internal=false)
...

2.2.3. 删除 Topic

public static void deleteTopics(AdminClient adminClient, String topicName) throws ExecutionException, InterruptedException {DeleteTopicsResult ret = adminClient.deleteTopics(Arrays.asList(topicName));ret.all().get();
}

2.2.4. 描述 Topic 详细信息

    一个 Topic 会有自身的描述信息，例如：partition 的数量，副本集的数量，是否为 internal等等。AdminClient 中提供了 describeTopics 方法来查询这些描述信息。代码示例：

public static void describeTopics(AdminClient adminClient, String topicName) throws ExecutionException, InterruptedException {DescribeTopicsResult ret = adminClient.describeTopics(Arrays.asList(topicName, "__consumer_offsets"));//等待返回结果完成Map<String, TopicDescription> topics = ret.all().get();for (Map.Entry<String, TopicDescription> entry : topics.entrySet()) {System.out.println(entry.getKey() + "---->" + entry.getValue());}}// 运行结果：
testTopic---->(name=testTopic, internal=false, partitions=(partition=0, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)), authorizedOperations=null)
__consumer_offsets---->(name=__consumer_offsets, internal=true, partitions=(partition=0, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=1, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=2, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=3, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=4, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=5, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=6, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=7, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=8, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=9, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=10, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=11, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=12, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=13, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=14, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=15, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=16, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=17, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=18, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=19, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=20, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=21, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=22, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=23, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=24, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=25, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=26, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=27, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=28, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=29, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=30, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=31, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=32, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=33, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=34, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=35, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=36, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=37, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=38, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=39, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=40, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=41, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=42, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=43, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=44, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=45, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=46, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=47, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=48, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)),(partition=49, leader=localhost:9092 (id: 0 rack: null), replicas=localhost:9092 (id: 0 rack: null), isr=localhost:9092 (id: 0 rack: null)), authorizedOperations=null)

2.3. 分区 Partition 操作

2.3.1. 增加分区

    在创建 Topic 时我们需要设定 Partition 的数量，但如果觉得初始设置的 Partition 数量太少了，那么就可以使用 createPartitions 方法来调整 Topic 的 Partition 数量，但是需要注意在 Kafka 中Partition 只能增加不能减少。代码示例：

public static void incrPartitions(AdminClient adminClient, String topicName) throws ExecutionException, InterruptedException {Map<String, NewPartitions> newPartitions = new HashMap<>();//将Partition数量调整为2newPartitions.put(topicName, NewPartitions.increaseTo(2));CreatePartitionsResult ret = adminClient.createPartitions(newPartitions);ret.all().get();
}

2.3.2. 分区副本重新分配

public void alterPartitionReassignments() throws ExecutionException, InterruptedException {// 构造要修改的topic和对应的partition idTopicPartition topicPartition = new TopicPartition("java_kafka_tst1", 1);//构造目标副本的partition id，其实就是broker idList<Integer> targetReplicas = new ArrayList<>();//把partition id添加进去targetReplicas.add(1);//使用targetReplicas构造newPartitionReassignmentNewPartitionReassignment newPartitionReassignment=new NewPartitionReassignment(targetReplicas);//构造重新分配的MapMap<TopicPartition, Optional<NewPartitionReassignment>> reassignments = new HashMap<>();//传入构造好的参数topicPartition和newPartitionReassignmentreassignments.put(topicPartition, Optional.of(newPartitionReassignment));//调用重新分配方法AlterPartitionReassignmentsResult result = adminClient.alterPartitionReassignments(reassignments);// 执行方法result.all().get();// 等待结果if (result.all().isDone()) {System.out.println("done");} else {System.out.println("not done");}
}

2.3.3. 查询分区副本列表

public static void listPartitionReassignments(AdminClient adminClient, String topicName) throws InterruptedException, ExecutionException {Set<TopicPartition> tpSet = new HashSet<>();tpSet.add(new TopicPartition("t-test", 0));ListPartitionReassignmentsResult ret = adminClient.listPartitionReassignments(tpSet);ret.reassignments().get();
}

2.4. Config 配置信息

2.4.1. 描述配置详细信息

public static void describeConfig(AdminClient adminClient, String topicName) throws ExecutionException, InterruptedException {DescribeConfigsResult ret = adminClient.describeConfigs(Collections.singleton(new ConfigResource(ConfigResource.Type.TOPIC, topicName)));Map<ConfigResource, Config> configMap = ret.all().get();for (Map.Entry<ConfigResource, Config> entry : configMap.entrySet()) {ConfigResource key = entry.getKey();Config value = entry.getValue();System.out.println(String.format("Resource type: %s, resource name: %s", key.type(), key.name()));Collection<ConfigEntry> configEntries = value.entries();for (ConfigEntry each : configEntries) {System.out.println(each.name() + " = " + each.value());}}
}// 运行结果：
Resource type: TOPIC, resource name: testTopic
compression.type = producer
leader.replication.throttled.replicas = 
message.downconversion.enable = true
min.insync.replicas = 1
segment.jitter.ms = 0
cleanup.policy = delete
flush.ms = 9223372036854775807
follower.replication.throttled.replicas = 
segment.bytes = 1073741824
retention.ms = 604800000
flush.messages = 9223372036854775807
message.format.version = 3.0-IV1
file.delete.delay.ms = 60000
max.compaction.lag.ms = 9223372036854775807
max.message.bytes = 1048588
min.compaction.lag.ms = 0
message.timestamp.type = CreateTime
preallocate = false
min.cleanable.dirty.ratio = 0.5
index.interval.bytes = 4096
unclean.leader.election.enable = false
retention.bytes = -1
delete.retention.ms = 86400000
segment.ms = 604800000
message.timestamp.difference.max.ms = 9223372036854775807
segment.index.bytes = 10485760

2.4.2. 修改配置信息

在早期版本中，使用 alterConfigs 方法来修改配置项。代码示例：

public static void alterConfigs(AdminClient adminClient, String topicName) throws ExecutionException, InterruptedException {// 指定ConfigResource的类型及名称ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, topicName);// 配置项以ConfigEntry形式存在Config config = new Config(Arrays.asList(new ConfigEntry("cleanup.policy", "compact")));//构建Map<ConfigResource, Config> configMap = new HashMap<>();configMap.put(configResource, config);//执行AlterConfigsResult ret = adminClient.alterConfigs(configMap);ret.all().get();
}

2.4.3. incrementalAlterConfigs

    在新版本中则是使用 incrementalAlterConfigs 方法来修改配置项，该方法使用起来相对于 alterConfigs 要略微复杂一些，但因此功能更多、更灵活。代码示例：

public static void incrementalAlterConfigs(AdminClient adminClient, String topicName) throws ExecutionException, InterruptedException {// 指定ConfigResource的类型及名称ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, topicName);// 配置项同样以ConfigEntry形式存在，只不过增加了操作类型// 以及能够支持操作多个配置项，相对来说功能更多、更灵活Collection<AlterConfigOp> configs = Arrays.asList(new AlterConfigOp(new ConfigEntry("preallocate", "false"),AlterConfigOp.OpType.SET));Map<ConfigResource, Collection<AlterConfigOp>> configMaps = new HashMap<>();configMaps.put(configResource, configs);AlterConfigsResult result = adminClient.incrementalAlterConfigs(configMaps);result.all().get();
}

2.5. Scram 账户操作

2.5.1. 创建 Scram 账户

public void createAccount(String name, String pwd, String salt) throws ExecutionException, InterruptedException {//创建User列表List<UserScramCredentialAlteration> alterations = new ArrayList<>();//构造Scram认证机制信息，这里笔者选择了SCRAM_SHA_512，大家也可以选择 ScramMechanism.SCRAM_SHA_256//alterations.size()此时为0，或许会报错，可以试下传入数字构造，比如下面添加了一个认证信息，那么这里传入数字1。// ScramCredentialInfo info=new ScramCredentialInfo(ScramMechanism.SCRAM_SHA_512, alterations.size()); //这里时间久远，忘记当时写例子的场景了ScramCredentialInfo info=new ScramCredentialInfo(ScramMechanism.SCRAM_SHA_512, 10000);//三个UserScramCredentialAlteration构造方法，三选一笔者选了一个最简单的//UserScramCredentialAlteration userScramCredentialAdd=new UserScramCredentialUpsertion(name,info,pwd.getBytes());//UserScramCredentialAlteration userScramCredentialAdd=new UserScramCredentialUpsertion(name,info,pwd.getBytes(),salt.getBytes());UserScramCredentialAlteration userScramCredentialAdd=new UserScramCredentialUpsertion(name,info,pwd);//添加认证信息到列表alterations.add(userScramCredentialAdd);//执行方法，并拿到返回结果AlterUserScramCredentialsResult result = adminClient.alterUserScramCredentials(alterations);//阻塞等待结果完成result.all().get();
}

2.5.2. 删除 Scram 账户

public void deleteAccount(String name) throws ExecutionException, InterruptedException {//创建删除列表List<UserScramCredentialAlteration> alterations = new ArrayList<>();//构建删除用的UserScramCredentialAlterationUserScramCredentialAlteration userScramCredentialDel=new UserScramCredentialDeletion(name,ScramMechanism.SCRAM_SHA_512);//添加认证信息到列表alterations.add(userScramCredentialDel);//执行方法，并拿到返回结果AlterUserScramCredentialsResult result = adminClient.alterUserScramCredentials(alterations);//阻塞等待结果完成result.all().get();
}

2.5.3. 查询 Scram 账户信息

public void describeAccount() throws ExecutionException, InterruptedException {//***************************************查询所有用户信息*****************************************************//查询所有的账户，这也是默认方法DescribeUserScramCredentialsResult result = adminClient.describeUserScramCredentials();//执行方法，并拿到返回结果Map<String, UserScramCredentialsDescription> future = result.all().get();//输出future.forEach((name,info)-> System.out.println("[ScramUserName:"+name+"]:[ScramUserInfo:"+info.toString()+"]"));//***************************************这里是分割线*****************************************************//***************************************查询指定的用户信息*****************************************************//构造指定的用户列表List<String> userScramList=new ArrayList<>();//添加两个用户userScramList.add("user1");userScramList.add("user2");//传入特定用户列表执行方法，并拿到返回结果DescribeUserScramCredentialsResult targetResult = adminClient.describeUserScramCredentials(userScramList);//执行方法，并拿到返回结果Map<String, UserScramCredentialsDescription> targetFuture = targetResult.all().get();//输出targetFuture.forEach((name,info)-> System.out.println("[ScramUserName:"+name+"]:[ScramUserInfo:"+info.toString()+"]"));
}

2.6. Acl 操作

2.6.1. 创建 Acl

public void createACL() {//创建ResourcePatternResourcePattern resourcePattern = new ResourcePattern(ResourceType.TOPIC, "topicName", PatternType.LITERAL);//创建AccessControlEntryAccessControlEntry accessControlEntry=new AccessControlEntry("User:accountName","*", AclOperation.READ, AclPermissionType.ALLOW);//绑定到AclBinding上AclBinding aclBinding=new AclBinding(resourcePattern,accessControlEntry);Collection<AclBinding> aclBindingCollection= new ArrayList<>();aclBindingCollection.add(aclBinding); //添加到集合CreateAclsResult aclResult = adminClient.createAcls(aclBindingCollection);KafkaFuture<Void> result = aclResult.all();try {result.get(); //执行if (result.isDone()){ //验证是否成功System.out.println(result.toString());}} catch (InterruptedException e) {e.printStackTrace();} catch (ExecutionException e) {e.printStackTrace();}
}

2.6.2. 删除 Acl

public void deleteAcls() {ResourcePatternFilter resourcePatternFilter = new ResourcePatternFilter(ResourceType.TOPIC, "topicName", PatternType.LITERAL);AccessControlEntryFilter accessControlEntryFilter=new AccessControlEntryFilter("User:accountName","*", AclOperation.ALL, AclPermissionType.ALLOW);AclBindingFilter aclBinding=new AclBindingFilter(resourcePatternFilter,accessControlEntryFilter);Collection<AclBindingFilter> aclBindingCollection= new ArrayList<>();aclBindingCollection.add(aclBinding);DeleteAclsResult aclResult = adminClient.deleteAcls(aclBindingCollection);KafkaFuture<Collection<AclBinding>> result = aclResult.all();try {result.get();if (result.isDone()){System.out.println(result.toString());}} catch (InterruptedException e) {e.printStackTrace();} catch (ExecutionException e) {e.printStackTrace();}
}