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);
    }
@Test
    public void testNoBrokerAclAuthorizer() throws Exception {
        Admin sourceAdmin = mock(Admin.class);
        Admin targetAdmin = mock(Admin.class);
        MirrorSourceConnector connector = new MirrorSourceConnector(sourceAdmin, targetAdmin,
                new MirrorSourceConfig(makeProps()));

        ExecutionException describeAclsFailure = new ExecutionException(
                "Failed to describe ACLs",
                new SecurityDisabledException("No ACL authorizer configured on this broker")
        );
        @SuppressWarnings("unchecked")
        KafkaFuture<Collection<AclBinding>> describeAclsFuture = mock(KafkaFuture.class);
        when(describeAclsFuture.get()).thenThrow(describeAclsFailure);
        DescribeAclsResult describeAclsResult = mock(DescribeAclsResult.class);
        when(describeAclsResult.values()).thenReturn(describeAclsFuture);
        when(sourceAdmin.describeAcls(any())).thenReturn(describeAclsResult);

        try (LogCaptureAppender connectorLogs = LogCaptureAppender.createAndRegister(MirrorSourceConnector.class)) {
            connectorLogs.setClassLogger(MirrorSourceConnector.class, Level.TRACE);
            connector.syncTopicAcls();
            long aclSyncDisableMessages = connectorLogs.getMessages().stream()
                    .filter(m -> m.contains("Consider disabling topic ACL syncing"))
                    .count();
            assertEquals(1, aclSyncDisableMessages, "Should have recommended that user disable ACL syncing");
            long aclSyncSkippingMessages = connectorLogs.getMessages().stream()
                    .filter(m -> m.contains("skipping topic ACL sync"))
                    .count();
            assertEquals(0, aclSyncSkippingMessages, "Should not have logged ACL sync skip at same time as suggesting ACL sync be disabled");

            connector.syncTopicAcls();
            connector.syncTopicAcls();
            aclSyncDisableMessages = connectorLogs.getMessages().stream()
                    .filter(m -> m.contains("Consider disabling topic ACL syncing"))
                    .count();
            assertEquals(1, aclSyncDisableMessages, "Should not have recommended that user disable ACL syncing more than once");
            aclSyncSkippingMessages = connectorLogs.getMessages().stream()
                    .filter(m -> m.contains("skipping topic ACL sync"))
                    .count();
            assertEquals(2, aclSyncSkippingMessages, "Should have logged ACL sync skip instead of suggesting disabling ACL syncing");
        }

        // We should never have tried to perform an ACL sync on the target cluster
        verifyNoInteractions(targetAdmin);
    }