@Override
    protected void doHealthCheck(Builder builder) {
        try (AdminClient adminClient = AdminClient.create(this.kafkaAdmin.getConfig())) {
            DescribeClusterResult result = adminClient.describeCluster(this.describeOptions);
            String brokerId = result.controller().get().idString();
            int replicationFactor = getReplicationFactor(brokerId, adminClient);
            int nodes = result.nodes().get().size();
            Health h = Option.when(nodes >= replicationFactor, builder::up)
                             .getOrElse(() ->
                                 builder.down()
                                        .withDetail("clusterId", result.clusterId())
                                        .withDetail("brokerId", brokerId)
                                        .withDetail("nodes", nodes))
                             .build();
            log.info("Current state kafka: {}", h.getStatus(), keyValue(HEALTH, h.getStatus()));
        } catch (Exception e) {
            Health h = builder.down().build();
            log.error("Current state kafka: {}, with error {}", h.getStatus(), e.toString(),
                keyValue(HEALTH, h.getStatus()));
        }
    }

AclBindingFilter filter = new AclBindingFilter(
new ResourcePatternFilter(ResourceType.ANY, null, PatternType.LITERAL),
new AccessControlEntryFilter(null, null, AclOperation.ANY, AclPermissionType.ANY));
adminClient.describeAcls(filter).values().get();

    /**
     * authorized operations for this topic, or null if this is not known.
     */
    public Set<AclOperation>  authorizedOperations() {
        return authorizedOperations;
    }

/**
 * Represents an operation which an ACL grants or denies permission to perform.
 *
 * Some operations imply other operations:
 * <ul>
 * <li><code>ALLOW ALL</code> implies <code>ALLOW</code> everything
 * <li><code>DENY ALL</code> implies <code>DENY</code> everything
 *
 * <li><code>ALLOW READ</code> implies <code>ALLOW DESCRIBE</code>
 * <li><code>ALLOW WRITE</code> implies <code>ALLOW DESCRIBE</code>
 * <li><code>ALLOW DELETE</code> implies <code>ALLOW DESCRIBE</code>
 *
 * <li><code>ALLOW ALTER</code> implies <code>ALLOW DESCRIBE</code>
 *
 * <li><code>ALLOW ALTER_CONFIGS</code> implies <code>ALLOW DESCRIBE_CONFIGS</code>
 * </ul>
 * The API for this class is still evolving and we may break compatibility in minor releases, if necessary.
 */
@InterfaceStability.Evolving
public enum AclOperation {