public Map<String, String> workerConfig(SourceAndTarget sourceAndTarget) {
        Map<String, String> props = new HashMap<>();
        props.putAll(clusterProps(sourceAndTarget.target()));

        // Accept common top-level configs that are otherwise ignored by MM2.
        // N.B. all other worker properties should be configured for specific herders,
        // e.g. primary->backup.client.id
        props.putAll(stringsWithPrefix("offset.storage"));
        props.putAll(stringsWithPrefix("config.storage"));
        props.putAll(stringsWithPrefix("status.storage"));
        props.putAll(stringsWithPrefix("key.converter")); 
        props.putAll(stringsWithPrefix("value.converter")); 
        props.putAll(stringsWithPrefix("header.converter"));
        props.putAll(stringsWithPrefix("task"));
        props.putAll(stringsWithPrefix("worker"));
        props.putAll(stringsWithPrefix("replication.policy"));

        // transform any expression like ${provider:path:key}, since the worker doesn't do so
        props = transform(props);
        props.putAll(stringsWithPrefix(CONFIG_PROVIDERS_CONFIG));

        // fill in reasonable defaults
        props.putIfAbsent(CommonClientConfigs.CLIENT_ID_CONFIG, sourceAndTarget.toString());
        props.putIfAbsent(GROUP_ID_CONFIG, sourceAndTarget.source() + "-mm2");
        props.putIfAbsent(DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG, "mm2-offsets."
                + sourceAndTarget.source() + ".internal");
        props.putIfAbsent(DistributedConfig.STATUS_STORAGE_TOPIC_CONFIG, "mm2-status."
                + sourceAndTarget.source() + ".internal");
        props.putIfAbsent(DistributedConfig.CONFIG_TOPIC_CONFIG, "mm2-configs."
                + sourceAndTarget.source() + ".internal");
        props.putIfAbsent(KEY_CONVERTER_CLASS_CONFIG, BYTE_ARRAY_CONVERTER_CLASS); 
        props.putIfAbsent(VALUE_CONVERTER_CLASS_CONFIG, BYTE_ARRAY_CONVERTER_CLASS); 
        props.putIfAbsent(HEADER_CONVERTER_CLASS_CONFIG, BYTE_ARRAY_CONVERTER_CLASS);

        return props;
    }
@Test
    public void testWorkerConfigs() {
        MirrorMakerConfig mirrorConfig = new MirrorMakerConfig(makeProps(
            "clusters", "a, b",
            "config.providers", "fake",
            "config.providers.fake.class", FakeConfigProvider.class.getName(),
            "replication.policy.separator", "__",
            "offset.storage.replication.factor", "123",
            "b.status.storage.replication.factor", "456",
            "b.producer.client.id", "client-one",
            "b.security.protocol", "PLAINTEXT",
            "b.producer.security.protocol", "SASL",
            "ssl.truststore.password", "secret1",
            "ssl.key.password", "${fake:secret:password}",  // resolves to "secret2"
            "b.xxx", "yyy"));
        SourceAndTarget a = new SourceAndTarget("b", "a");
        SourceAndTarget b = new SourceAndTarget("a", "b");
        Map<String, String> aProps = mirrorConfig.workerConfig(a);
        assertEquals("b->a", aProps.get("client.id"));
        assertEquals("123", aProps.get("offset.storage.replication.factor"));
        assertEquals("__", aProps.get("replication.policy.separator"));
        assertEquals("fake", aProps.get("config.providers"));
        Map<String, String> bProps = mirrorConfig.workerConfig(b);
        assertEquals("a->b", bProps.get("client.id"));
        assertEquals("456", bProps.get("status.storage.replication.factor"));
        assertEquals("client-one", bProps.get("producer.client.id"),
            "producer props should be passed through to worker producer config: " + bProps);
        assertEquals("SASL", bProps.get("producer.security.protocol"),
            "replication-level security props should be passed through to worker producer config");
        assertEquals("SASL", bProps.get("producer.security.protocol"),
            "replication-level security props should be passed through to worker producer config");
        assertEquals("PLAINTEXT", bProps.get("consumer.security.protocol"),
            "replication-level security props should be passed through to worker consumer config");
        assertEquals("secret1", bProps.get("ssl.truststore.password"),
            "security properties should be passed through to worker config: " + bProps);
        assertEquals("secret1", bProps.get("producer.ssl.truststore.password"),
            "security properties should be passed through to worker producer config: " + bProps);
        assertEquals("secret2", bProps.get("ssl.key.password"),
            "security properties should be transformed in worker config");
        assertEquals("secret2", bProps.get("producer.ssl.key.password"),
            "security properties should be transformed in worker producer config");
        assertEquals("__", bProps.get("replication.policy.separator"));
    }