private Optional<Collection<AclBinding>> listTopicAclBindings()
            throws InterruptedException, ExecutionException {
        return adminCall(
                () -> {
                    Collection<AclBinding> bindings;
                    try {
                        bindings = sourceAdminClient.describeAcls(ANY_TOPIC_ACL).values().get();
                    } catch (ExecutionException e) {
                        if (e.getCause() instanceof SecurityDisabledException) {
                            if (noAclAuthorizer.compareAndSet(false, true)) {
                                log.info(
                                        "No ACL authorizer is configured on the source Kafka cluster, so no topic ACL syncing will take place. "
                                                + "Consider disabling topic ACL syncing by setting " + SYNC_TOPIC_ACLS_ENABLED + " to 'false'."
                                );
                            } else {
                                log.debug("Source-side ACL authorizer still not found; skipping topic ACL sync");
                            }
                            return Optional.empty();
                        } else {
                            throw e;
                        }
                    }
                    return Optional.of(bindings);
                },
                () -> "describe ACLs on " + config.sourceClusterAlias() + " cluster"
        );
    }
Replicate data, configuration, and ACLs between clusters.

 @see MirrorSourceConfig for supported config properties.
connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorSourceConnector.java