@BeforeEach
    public void startClusters() throws Exception {
        Map<String, Object> sslConfig = TestSslUtils.createSslConfig(false, true, ConnectionMode.SERVER, TestUtils.tempFile(), "testCert");
        // enable SSL on backup kafka broker
        backupBrokerProps.put(SocketServerConfigs.LISTENER_SECURITY_PROTOCOL_MAP_CONFIG, "EXTERNAL:SSL,CONTROLLER:SSL");
        backupBrokerProps.putAll(sslConfig);

        Properties sslProps = new Properties();
        sslProps.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, sslConfig.get(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG));
        sslProps.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, ((Password) sslConfig.get(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG)).value());
        sslProps.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");

        // set SSL config for kafka connect worker
        backupWorkerProps.putAll(sslProps.entrySet().stream().collect(Collectors.toMap(
            e -> String.valueOf(e.getKey()), e ->  String.valueOf(e.getValue()))));

        mm2Props.putAll(sslProps.entrySet().stream().collect(Collectors.toMap(
            e -> BACKUP_CLUSTER_ALIAS + "." + e.getKey(), e ->  String.valueOf(e.getValue()))));
        // set SSL config for producer used by source task in MM2
        mm2Props.putAll(sslProps.entrySet().stream().collect(Collectors.toMap(
            e -> BACKUP_CLUSTER_ALIAS + ".producer." + e.getKey(), e ->  String.valueOf(e.getValue()))));

        super.startClusters();
    }
Tests MM2 replication with SSL enabled at backup kafka cluster
connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationSSLTest.java