@Override
    public CompletionStage<Set<String>> getAllUsers() {
        LOGGER.debugOp("Searching for Users with any ACL rules");

        Set<String> users = new HashSet<>();
        Set<String> ignored = new HashSet<>(IGNORED_USERS.size());
        Enumeration<String> keys = cache.keys();

        while (keys.hasMoreElements())  {
            String username = KafkaUserModel.decodeUsername(keys.nextElement());

            if (IGNORED_USERS.contains(username)) {
                if (!ignored.contains(username)) {
                    // This info message is logged only once per reconciliation even if there are multiple rules
                    LOGGER.infoOp("Existing ACLs for user '{}' will be ignored.", username);
                    ignored.add(username);
                }
            } else {
                if (LOGGER.isDebugEnabled()) {
                    LOGGER.debugOp("Adding user {} to Set of users with ACLs", username);
                }

                users.add(username);
            }
        }

        return CompletableFuture.completedFuture(users);
    }
@Test
    public void testGetAllUsers() throws ExecutionException, InterruptedException {
        Admin mockAdminClient = mock(AdminClient.class);

        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));

        SimpleAclOperator aclOp = new SimpleAclOperator(mockAdminClient, ResourceUtils.createUserOperatorConfig(), EXECUTOR);
        aclOp.start();

        try {
            Set<String> users = aclOp.getAllUsers().toCompletableFuture().get();
            assertThat(users, is(new HashSet<>(asList("foo", "bar", "baz"))));
        } finally {
            aclOp.stop();
        }
    }