public Map<String, String> connectorBaseConfig(SourceAndTarget sourceAndTarget, Class<?> connectorClass) {
        Map<String, String> props = new HashMap<>();

        props.putAll(rawProperties);
        props.keySet().retainAll(allConfigNames());
        
        props.putAll(stringsWithPrefix(CONFIG_PROVIDERS_CONFIG));
        props.putAll(stringsWithPrefix("replication.policy"));

        Map<String, String> sourceClusterProps = clusterProps(sourceAndTarget.source());
        // attrs non prefixed with producer|consumer|admin
        props.putAll(clusterConfigsWithPrefix(SOURCE_CLUSTER_PREFIX, sourceClusterProps));
        // attrs prefixed with producer|consumer|admin
        props.putAll(clientConfigsWithPrefix(SOURCE_PREFIX, sourceClusterProps));

        Map<String, String> targetClusterProps = clusterProps(sourceAndTarget.target());
        props.putAll(clusterConfigsWithPrefix(TARGET_CLUSTER_PREFIX, targetClusterProps));
        props.putAll(clientConfigsWithPrefix(TARGET_PREFIX, targetClusterProps));

        props.putIfAbsent(NAME, connectorClass.getSimpleName());
        props.putIfAbsent(CONNECTOR_CLASS, connectorClass.getName());
        props.putIfAbsent(SOURCE_CLUSTER_ALIAS, sourceAndTarget.source());
        props.putIfAbsent(TARGET_CLUSTER_ALIAS, sourceAndTarget.target());

        // override with connector-level properties
        props.putAll(stringsWithPrefixStripped(sourceAndTarget.source() + "->"
            + sourceAndTarget.target() + "."));

        // disabled by default
        props.putIfAbsent(MirrorConnectorConfig.ENABLED, "false");

        // don't transform -- the worker will handle transformation of Connector and Task configs
        return props;
    }
@Test
    public void testClusterConfigProperties() {
        MirrorMakerConfig mirrorConfig = new MirrorMakerConfig(makeProps(
            "clusters", "a, b",
            "a.bootstrap.servers", "servers-one",
            "b.bootstrap.servers", "servers-two",
            "security.protocol", "SSL",
            "replication.factor", "4"));
        Map<String, String> connectorProps = mirrorConfig.connectorBaseConfig(new SourceAndTarget("a", "b"),
            MirrorSourceConnector.class);
        assertEquals("servers-one", connectorProps.get("source.cluster.bootstrap.servers"),
            "source.cluster.bootstrap.servers is set");
        assertEquals("servers-two", connectorProps.get("target.cluster.bootstrap.servers"),
            "target.cluster.bootstrap.servers is set");
        assertEquals("SSL", connectorProps.get("security.protocol"),
            "top-level security.protocol is passed through to connector config");
    }