@Test
    public void testCreatePartitionsUseProvidedForwardingAdmin() throws Exception {
        mm2Config = new MirrorMakerConfig(mm2Props);
        produceMessages(backupProducer, "test-topic-1");
        produceMessages(primaryProducer, "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);

        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");

        // make sure to remote topics are created using FakeForwardingAdminWithLocalMetadata
        waitForTopicToPersistInFakeLocalMetadataStore("backup.test-topic-1");
        waitForTopicToPersistInFakeLocalMetadataStore("primary.test-topic-1");

        // increase number of partitions
        Map<String, NewPartitions> newPartitions = Collections.singletonMap("test-topic-1", NewPartitions.increaseTo(NUM_PARTITIONS + 1));
        try (Admin adminClient = primary.kafka().createAdminClient()) {
            adminClient.createPartitions(newPartitions).all().get();
        }

        // make sure partition is created in the other cluster
        waitForTopicPartitionCreated(backup, "primary.test-topic-1", NUM_PARTITIONS + 1);

        // expect to use FakeForwardingAdminWithLocalMetadata to update number of partitions in local store
        waitForTopicConfigPersistInFakeLocalMetaDataStore("primary.test-topic-1", "partitions", String.valueOf(NUM_PARTITIONS + 1));
    }
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