Try based on AdminClient define ACL in Kafka:
KafkaPrincipal principal = new KafkaPrincipal("User", "test-producer");
ResourcePattern resourcePattern = new ResourcePattern(ResourceType.TOPIC, "test", PatternType.LITERAL);
AclBinding aclBinding = new AclBinding(resourcePattern, new AccessControlEntry(principal.toString(), "*", AclOperation.WRITE, AclPermissionType.ALLOW));
CreateAclsResult result = admin.createAcls(Collections.singletonList(aclBinding));
result.all().get();
in file server.properties add one property: (based on doc)
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
when run code get error:
java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.SecurityDisabledException: No Authorizer is configured. at java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:396) at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2073) at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:165) at com.comarch.clm6.lib.businessConfiguration.PrepareACL.main(PrepareACL.java:31) Caused by: org.apache.kafka.common.errors.SecurityDisabledException: No Authorizer is configured.