java kafka如何动态设置用户读写权限

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.DescribeAclsResult;
import org.apache.kafka.common.acl.AccessControlEntry;
import org.apache.kafka.common.acl.AclBinding;
import org.apache.kafka.common.acl.AclBindingFilter;
import org.apache.kafka.common.acl.AclOperation;
import org.apache.kafka.common.acl.AclPermissionType;
import org.apache.kafka.common.resource.PatternType;
import org.apache.kafka.common.resource.ResourcePattern;
import org.apache.kafka.common.resource.ResourceType;
import org.springframework.kafka.core.KafkaAdmin;
public static void main(String[] args) {
        Map<String, Object> configs = new HashMap<>();
        // broker地址，多个用逗号分割
        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "ip:端口");
        configs.put("security.protocol", "SASL_PLAINTEXT");
        configs.put("sasl.mechanism", "SCRAM-SHA-512");
        // 登录broker的账户 admin是管理员
        configs.put("sasl.jaas.config",
            "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"admin\" password=\"admin\";");
        KafkaAdmin admin = new KafkaAdmin(configs);
        AdminClient adminClient = AdminClient.create(admin.getConfigurationProperties());
        // principal：User:test2是需要赋予权限的帐号
        // host：主机 （*号即可）
        // operation：权限操作
        // permissionType：权限类型
        AccessControlEntry ace = new AccessControlEntry("User:test2", "*", AclOperation.WRITE, AclPermissionType.ALLOW);
        // resourceType：资源类型（topic）
        // name：topic名称
        // patternType：资源模式类型
        ResourcePattern rp = new ResourcePattern(ResourceType.TOPIC, "Z7TEST", PatternType.LITERAL);
        AclBinding ab = new AclBinding(rp, ace);
        // 多个权限赋予可以传list
        List<AclBinding> ablist = Arrays.asList(ab);
        adminClient.createAcls(ablist);
        // 可以查看赋予用户的所有权限
        DescribeAclsResult b = adminClient.describeAcls(AclBindingFilter.ANY);
        System.out.println(b.values());
        adminClient.close();
    }