@Test
    public void testReplicationIsCreatingTopicsUsingProvidedForwardingAdmin() throws Exception {
        produceMessages(primaryProducer, "test-topic-1");
        produceMessages(backupProducer, "test-topic-1");
        String consumerGroupName = "consumer-group-testReplication";
        Map<String, Object> consumerProps = Collections.singletonMap("group.id", consumerGroupName);
        // warm up consumers before starting the connectors so we don't need to wait for discovery
        warmUpConsumer(consumerProps);

        mm2Config = new MirrorMakerConfig(mm2Props);

        waitUntilMirrorMakerIsRunning(backup, CONNECTOR_LIST, mm2Config, PRIMARY_CLUSTER_ALIAS, BACKUP_CLUSTER_ALIAS);
        waitUntilMirrorMakerIsRunning(primary, CONNECTOR_LIST, mm2Config, BACKUP_CLUSTER_ALIAS, PRIMARY_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");
        waitForTopicCreated(primary, "mm2-offset-syncs.backup.internal");
        waitForTopicCreated(primary, "backup.checkpoints.internal");
        waitForTopicCreated(primary, "backup.heartbeats");
        waitForTopicCreated(backup, "mm2-offset-syncs.primary.internal");
        waitForTopicCreated(backup, "primary.checkpoints.internal");
        waitForTopicCreated(backup, "primary.heartbeats");

        // expect to use FakeForwardingAdminWithLocalMetadata to create remote topics and internal topics into local store
        waitForTopicToPersistInFakeLocalMetadataStore("backup.test-topic-1");
        waitForTopicToPersistInFakeLocalMetadataStore("primary.test-topic-1");
        waitForTopicToPersistInFakeLocalMetadataStore("mm2-offset-syncs.backup.internal");
        waitForTopicToPersistInFakeLocalMetadataStore("backup.checkpoints.internal");
        waitForTopicToPersistInFakeLocalMetadataStore("backup.heartbeats");
        waitForTopicToPersistInFakeLocalMetadataStore("mm2-offset-syncs.primary.internal");
        waitForTopicToPersistInFakeLocalMetadataStore("primary.checkpoints.internal");
        waitForTopicToPersistInFakeLocalMetadataStore("primary.heartbeats");
        waitForTopicToPersistInFakeLocalMetadataStore("heartbeats");
    }
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