public MirrorClientConfig clientConfig(String cluster) {
        Map<String, String> props = new HashMap<>();
        props.putAll(originalsStrings());
        props.putAll(clusterProps(cluster));
        return new MirrorClientConfig(transform(props));
    }
@Test
    public void testClientConfigProperties() {
        String clusterABootstrap = "127.0.0.1:9092, 127.0.0.2:9092";
        String clusterBBootstrap = "127.0.0.3:9092, 127.0.0.4:9092";
        MirrorMakerConfig mirrorConfig = new MirrorMakerConfig(makeProps(
            "clusters", "a, b",
            "config.providers", "fake",
            "config.providers.fake.class", FakeConfigProvider.class.getName(),
            "replication.policy.separator", "__",
            "ssl.key.password", "${fake:secret:password}",  // resolves to "secret2"
            "security.protocol", "SSL",
            "a.security.protocol", "PLAINTEXT",
            "a.producer.security.protocol", "SSL",
            "a.bootstrap.servers", clusterABootstrap,
            "b.bootstrap.servers", clusterBBootstrap,
            "metrics.reporter", FakeMetricsReporter.class.getName(),
            "a.metrics.reporter", FakeMetricsReporter.class.getName(),
            "b->a.metrics.reporter", FakeMetricsReporter.class.getName(),
            "b.forwarding.admin.class", FakeForwardingAdmin.class.getName(),
            "a.xxx", "yyy",
            "xxx", "zzz"));
        MirrorClientConfig aClientConfig = mirrorConfig.clientConfig("a");
        MirrorClientConfig bClientConfig = mirrorConfig.clientConfig("b");
        assertEquals("__", aClientConfig.getString("replication.policy.separator"),
            "replication.policy.separator is picked up in MirrorClientConfig");
        assertEquals("b__topic1", aClientConfig.replicationPolicy().formatRemoteTopic("b", "topic1"),
            "replication.policy.separator is honored");
        assertEquals(clusterABootstrap, aClientConfig.adminConfig().get("bootstrap.servers"),
            "client configs include bootstrap.servers");
        try (ForwardingAdmin forwardingAdmin = aClientConfig.forwardingAdmin(aClientConfig.adminConfig())) {
            assertEquals(ForwardingAdmin.class.getName(), forwardingAdmin.getClass().getName(),
                    "Cluster a uses the default ForwardingAdmin");
        }
        assertEquals("PLAINTEXT", aClientConfig.adminConfig().get("security.protocol"),
            "client configs include security.protocol");
        assertEquals("SSL", aClientConfig.producerConfig().get("security.protocol"),
            "producer configs include security.protocol");
        assertFalse(aClientConfig.adminConfig().containsKey("xxx"),
            "unknown properties aren't included in client configs");
        assertFalse(aClientConfig.adminConfig().containsKey("metric.reporters"),
            "top-level metrics reporters aren't included in client configs");
        assertEquals("secret2", aClientConfig.getPassword("ssl.key.password").value(),
            "security properties are translated from external sources");
        assertEquals("secret2", ((Password) aClientConfig.adminConfig().get("ssl.key.password")).value(),
            "client configs are translated from external sources");
        assertFalse(aClientConfig.producerConfig().containsKey("metrics.reporter"),
            "client configs should not include metrics reporter");
        assertFalse(bClientConfig.adminConfig().containsKey("metrics.reporter"),
            "client configs should not include metrics reporter");
        try (ForwardingAdmin forwardingAdmin = bClientConfig.forwardingAdmin(bClientConfig.adminConfig())) {
            assertEquals(FakeForwardingAdmin.class.getName(), forwardingAdmin.getClass().getName(),
                    "Cluster b should use the FakeForwardingAdmin");
        }
    }