@BeforeEach
    public void startClusters() throws Exception {
        enableAclAuthorizer(primaryBrokerProps);
        additionalPrimaryClusterClientsConfigs.putAll(superUserConfig());
        primaryWorkerProps.putAll(superUserConfig());

        enableAclAuthorizer(backupBrokerProps);
        additionalBackupClusterClientsConfigs.putAll(superUserConfig());
        backupWorkerProps.putAll(superUserConfig());

        HashMap<String, String> additionalConfig = new HashMap<String, String>(superUserConfig()) {{
                put(FORWARDING_ADMIN_CLASS, FakeForwardingAdminWithLocalMetadata.class.getName());
            }};

        superUserConfig().forEach((property, value) -> {
            additionalConfig.put(CONNECTOR_CLIENT_CONSUMER_OVERRIDES_PREFIX + property, value);
            additionalConfig.put(CONNECTOR_CLIENT_PRODUCER_OVERRIDES_PREFIX + property, value);
            additionalConfig.put("consumer." + property, value);
            additionalConfig.put("producer." + property, value);
        });

        connectorUserConfig().forEach((property, value) -> {
            additionalConfig.put("admin." + property, value);
            additionalConfig.put(CONNECTOR_CLIENT_ADMIN_OVERRIDES_PREFIX + property, value);
        });

        startClusters(additionalConfig);

        try (Admin adminClient = primary.kafka().createAdminClient()) {
            adminClient.createAcls(Collections.singletonList(
                    new AclBinding(
                            new ResourcePattern(ResourceType.TOPIC, "*", PatternType.LITERAL),
                            new AccessControlEntry("User:connector", "*", AclOperation.ALL, AclPermissionType.ALLOW)
                    )
            )).all().get();
        }
        try (Admin adminClient = backup.kafka().createAdminClient()) {
            adminClient.createAcls(Collections.singletonList(
                    new AclBinding(
                            new ResourcePattern(ResourceType.TOPIC, "*", PatternType.LITERAL),
                            new AccessControlEntry("User:connector", "*", AclOperation.ALL, AclPermissionType.ALLOW)
                    )
            )).all().get();
        }
    }
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