Map<String, Object> sourceConsumerConfig(String role) {
        Map<String, Object> result = sourceConsumerConfig(originals());
        addClientId(result, role);
        return result;
    }
@Test
    public void testSourceConsumerConfig() {
        Map<String, String> connectorProps = makeProps(
                MirrorConnectorConfig.CONSUMER_CLIENT_PREFIX + "max.poll.interval.ms", "120000",
                MirrorConnectorConfig.SOURCE_CLUSTER_PREFIX + "bootstrap.servers", "localhost:2345"
        );
        MirrorConnectorConfig config = new TestMirrorConnectorConfig(connectorProps);
        Map<String, Object> connectorConsumerProps = config.sourceConsumerConfig("test");
        Map<String, Object> expectedConsumerProps = new HashMap<>();
        expectedConsumerProps.put("enable.auto.commit", "false");
        expectedConsumerProps.put("auto.offset.reset", "earliest");
        expectedConsumerProps.put("max.poll.interval.ms", "120000");
        expectedConsumerProps.put("client.id", "source1->target2|ConnectorName|test");
        expectedConsumerProps.put("bootstrap.servers", "localhost:2345");
        assertEquals(expectedConsumerProps, connectorConsumerProps);

        // checking auto.offset.reset override works
        connectorProps = makeProps(
                MirrorConnectorConfig.CONSUMER_CLIENT_PREFIX + "auto.offset.reset", "latest",
                MirrorConnectorConfig.SOURCE_CLUSTER_PREFIX + "bootstrap.servers", "localhost:2345"
        );
        config = new TestMirrorConnectorConfig(connectorProps);
        connectorConsumerProps = config.sourceConsumerConfig("test");
        expectedConsumerProps.put("auto.offset.reset", "latest");
        expectedConsumerProps.remove("max.poll.interval.ms");
        assertEquals(expectedConsumerProps, connectorConsumerProps,
                MirrorConnectorConfig.CONSUMER_CLIENT_PREFIX + " source consumer config not matching");
    }