private void updateTopicAcls(List<AclBinding> bindings) throws ExecutionException, InterruptedException {
        log.trace("Syncing {} topic ACL bindings.", bindings.size());
        adminCall(
                () -> {
                    targetAdminClient.createAcls(bindings).values().forEach((k, v) -> v.whenComplete((x, e) -> {
                        if (e != null) {
                            log.warn("Could not sync ACL of topic {}.", k.pattern().name(), e);
                        }
                    }));
                    return null;
                },
                () -> String.format("create ACLs %s on %s cluster", bindings, config.targetClusterAlias())
        );
    }
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