SimpleAclOperator { public Set<String> getUsersWithAcls() { Set<String> result = new HashSet<>(); Set<String> ignored = new HashSet<>(IGNORED_USERS.size()); log.debug("Searching for Users with any ACL rules"); Collection<AclBinding> aclBindings; try { aclBindings = adminClient.describeAcls(AclBindingFilter.ANY).values().get(); } catch (InterruptedException | ExecutionException e) { return result; } for (AclBinding aclBinding : aclBindings) { KafkaPrincipal principal = SecurityUtils.parseKafkaPrincipal(aclBinding.entry().principal()); if (KafkaPrincipal.USER_TYPE.equals(principal.getPrincipalType())) { String username = KafkaUserModel.decodeUsername(principal.getName()); if (IGNORED_USERS.contains(username)) { if (!ignored.contains(username)) { log.info("Existing ACLs for user '{}' will be ignored.", username); ignored.add(username); } } else { if (log.isTraceEnabled()) { log.trace("Adding user {} to Set of users with ACLs", username); } result.add(username); } } } return result; } SimpleAclOperator(Vertx vertx, Admin adminClient); Future<ReconcileResult<Set<SimpleAclRule>>> reconcile(String username, Set<SimpleAclRule> desired); Set<SimpleAclRule> getAcls(String username); Set<String> getUsersWithAcls();  }
@Test public void testGetUsersFromAcls(VertxTestContext context) { Admin mockAdminClient = mock(AdminClient.class); SimpleAclOperator aclOp = new SimpleAclOperator(vertx, mockAdminClient); ResourcePattern res1 = new ResourcePattern(ResourceType.TOPIC, "my-topic", PatternType.LITERAL); ResourcePattern res2 = new ResourcePattern(ResourceType.GROUP, "my-group", PatternType.LITERAL); KafkaPrincipal foo = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "CN=foo"); AclBinding fooAclBinding = new AclBinding(res1, new AccessControlEntry(foo.toString(), "*", org.apache.kafka.common.acl.AclOperation.READ, AclPermissionType.ALLOW)); KafkaPrincipal bar = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "CN=bar"); AclBinding barAclBinding = new AclBinding(res1, new AccessControlEntry(bar.toString(), "*", org.apache.kafka.common.acl.AclOperation.READ, AclPermissionType.ALLOW)); KafkaPrincipal baz = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "baz"); AclBinding bazAclBinding = new AclBinding(res2, new AccessControlEntry(baz.toString(), "*", org.apache.kafka.common.acl.AclOperation.READ, AclPermissionType.ALLOW)); KafkaPrincipal all = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "*"); AclBinding allAclBinding = new AclBinding(res1, new AccessControlEntry(all.toString(), "*", org.apache.kafka.common.acl.AclOperation.READ, AclPermissionType.ALLOW)); KafkaPrincipal anonymous = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "ANONYMOUS"); AclBinding anonymousAclBinding = new AclBinding(res2, new AccessControlEntry(anonymous.toString(), "*", org.apache.kafka.common.acl.AclOperation.READ, AclPermissionType.ALLOW)); Collection<AclBinding> aclBindings = asList(fooAclBinding, barAclBinding, bazAclBinding, allAclBinding, anonymousAclBinding); assertDoesNotThrow(() -> mockDescribeAcls(mockAdminClient, AclBindingFilter.ANY, aclBindings)); assertThat(aclOp.getUsersWithAcls(), is(new HashSet<>(asList("foo", "bar", "baz")))); context.completeNow(); }