protected static void enableAclAuthorizer(Properties brokerProps) {
        brokerProps.put(SocketServerConfigs.LISTENER_SECURITY_PROTOCOL_MAP_CONFIG, "CONTROLLER:SASL_PLAINTEXT,EXTERNAL:SASL_PLAINTEXT");
        brokerProps.put(ServerConfigs.AUTHORIZER_CLASS_NAME_CONFIG, "org.apache.kafka.metadata.authorizer.StandardAuthorizer");
        brokerProps.put(BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG, "PLAIN");
        brokerProps.put(BrokerSecurityConfigs.SASL_MECHANISM_INTER_BROKER_PROTOCOL_CONFIG, "PLAIN");
        brokerProps.put(KRaftConfigs.SASL_MECHANISM_CONTROLLER_PROTOCOL_CONFIG, "PLAIN");
        String listenerSaslJaasConfig = "org.apache.kafka.common.security.plain.PlainLoginModule required "
                + "username=\"super\" "
                + "password=\"super_pwd\" "
                + "user_connector=\"connector_pwd\" "
                + "user_super=\"super_pwd\";";
        brokerProps.put("listener.name.external.plain.sasl.jaas.config", listenerSaslJaasConfig);
        brokerProps.put("listener.name.controller.plain.sasl.jaas.config", listenerSaslJaasConfig);
        brokerProps.put("super.users", "User:super");
    }
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