@Override
    public synchronized void putConnectorConfig(String connector, Map<String, String> properties, TargetState targetState) {
        ConnectorState state = connectors.get(connector);
        if (state == null)
            connectors.put(connector, new ConnectorState(properties, targetState));
        else {
            state.connConfig = properties;
            if (targetState != null) {
                state.targetState = targetState;
            }
        }

        if (updateListener != null)
            updateListener.onConnectorConfigUpdate(connector);
    }
@Test
    public void testPutConnectorConfig() {
        configStore.putConnectorConfig(CONNECTOR_IDS.get(0), SAMPLE_CONFIGS.get(0), null);
        ClusterConfigState configState = configStore.snapshot();

        assertTrue(configState.contains(CONNECTOR_IDS.get(0)));
        // Default initial target state of STARTED should be used if no explicit target state is specified
        assertEquals(TargetState.STARTED, configState.targetState(CONNECTOR_IDS.get(0)));
        assertEquals(SAMPLE_CONFIGS.get(0), configState.connectorConfig(CONNECTOR_IDS.get(0)));
        assertEquals(1, configState.connectors().size());

        verify(configUpdateListener).onConnectorConfigUpdate(eq(CONNECTOR_IDS.get(0)));
    }