@Override
    public ConnectorInfo connectorInfo(String connector) {
        final ClusterConfigState configState = configBackingStore.snapshot();

        if (!configState.contains(connector))
            return null;
        Map<String, String> config = configState.rawConnectorConfig(connector);

        return new ConnectorInfo(
            connector,
            config,
            configState.tasks(connector),
            connectorType(config)
        );
    }
@Test
    public void testConnectorInfo() {
        AbstractHerder herder = testHerder();

        when(plugins.newConnector(anyString())).thenReturn(new SampleSourceConnector());
        when(herder.plugins()).thenReturn(plugins);

        when(configStore.snapshot()).thenReturn(SNAPSHOT);

        ConnectorInfo info = herder.connectorInfo(CONN1);

        assertEquals(CONN1, info.name());
        assertEquals(CONN1_CONFIG, info.config());
        assertEquals(Arrays.asList(TASK0, TASK1, TASK2), info.tasks());
        assertEquals(ConnectorType.SOURCE, info.type());
    }