@Test
    public void testSyncTopicACLsUseProvidedForwardingAdmin() throws Exception {
        mm2Props.put("sync.topic.acls.enabled", "true");
        mm2Props.put("sync.topic.acls.interval.seconds", "1");
        mm2Config = new MirrorMakerConfig(mm2Props);
        List<AclBinding> aclBindings = Collections.singletonList(
                new AclBinding(
                        new ResourcePattern(ResourceType.TOPIC, "test-topic-1", PatternType.LITERAL),
                        new AccessControlEntry("User:dummy", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW)
                )
        );
        try (Admin adminClient = primary.kafka().createAdminClient()) {
            adminClient.createAcls(aclBindings).all().get();
        }
        try (Admin adminClient = backup.kafka().createAdminClient()) {
            adminClient.createAcls(aclBindings).all().get();
        }

        waitUntilMirrorMakerIsRunning(primary, CONNECTOR_LIST, mm2Config, BACKUP_CLUSTER_ALIAS, PRIMARY_CLUSTER_ALIAS);
        waitUntilMirrorMakerIsRunning(backup, CONNECTOR_LIST, mm2Config, PRIMARY_CLUSTER_ALIAS, BACKUP_CLUSTER_ALIAS);

        // make sure the topic is auto-created in the other cluster
        waitForTopicCreated(primary, "backup.test-topic-1");
        waitForTopicCreated(backup, "primary.test-topic-1");

        // expect to use FakeForwardingAdminWithLocalMetadata to update topic ACLs on remote cluster
        AclBinding expectedACLOnBackupCluster = new AclBinding(
                new ResourcePattern(ResourceType.TOPIC, "primary.test-topic-1", PatternType.LITERAL),
                new AccessControlEntry("User:dummy", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW)
        );
        AclBinding expectedACLOnPrimaryCluster = new AclBinding(
                new ResourcePattern(ResourceType.TOPIC, "backup.test-topic-1", PatternType.LITERAL),
                new AccessControlEntry("User:dummy", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW)
        );

        // In some rare cases replica topics are created before ACLs are synced, so retry logic is necessary
        waitForCondition(
                () -> {
                    assertTrue(getAclBindings(backup.kafka(), "primary.test-topic-1").contains(expectedACLOnBackupCluster), "topic ACLs are not synced on backup cluster");
                    assertTrue(getAclBindings(primary.kafka(), "backup.test-topic-1").contains(expectedACLOnPrimaryCluster), "topic ACLs are not synced on primary cluster");
                    return true;
                },
                TOPIC_ACL_SYNC_DURATION_MS,
                "Topic ACLs were not synced in time"
        );

        // expect to use FakeForwardingAdminWithLocalMetadata to update topic ACLs in FakeLocalMetadataStore.allAcls
        assertTrue(FakeLocalMetadataStore.aclBindings("dummy").containsAll(Arrays.asList(expectedACLOnBackupCluster, expectedACLOnPrimaryCluster)));
    }
Tests MM2 is using provided ForwardingAdmin to create/alter topics, partitions and ACLs.
connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsWithCustomForwardingAdminIntegrationTest.java