void syncTopicAcls()
            throws InterruptedException, ExecutionException {
        Optional<Collection<AclBinding>> rawBindings = listTopicAclBindings();
        if (!rawBindings.isPresent())
            return;
        List<AclBinding> filteredBindings = rawBindings.get().stream()
            .filter(x -> x.pattern().resourceType() == ResourceType.TOPIC)
            .filter(x -> x.pattern().patternType() == PatternType.LITERAL)
            .filter(this::shouldReplicateAcl)
            .filter(x -> shouldReplicateTopic(x.pattern().name()))
            .map(this::targetAclBinding)
            .collect(Collectors.toList());
        updateTopicAcls(filteredBindings);
    }
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