@Override
    public synchronized void putTaskConfigs(String connector, List<Map<String, String>> configs) {
        ConnectorState state = connectors.get(connector);
        if (state == null)
            throw new IllegalArgumentException("Cannot put tasks for non-existing connector");

        Map<ConnectorTaskId, Map<String, String>> taskConfigsMap = taskConfigListAsMap(connector, configs);
        state.taskConfigs = taskConfigsMap;

        state.applyConfig();

        if (updateListener != null)
            updateListener.onTaskConfigUpdate(taskConfigsMap.keySet());
    }
@Test
    public void testPutTaskConfigs() {
        // Can't write task configs for non-existent connector
        assertThrows(IllegalArgumentException.class,
            () -> configStore.putTaskConfigs(CONNECTOR_IDS.get(0), Collections.singletonList(SAMPLE_CONFIGS.get(1))));

        configStore.putConnectorConfig(CONNECTOR_IDS.get(0), SAMPLE_CONFIGS.get(0), null);
        configStore.putTaskConfigs(CONNECTOR_IDS.get(0), Collections.singletonList(SAMPLE_CONFIGS.get(1)));
        ClusterConfigState configState = configStore.snapshot();

        ConnectorTaskId taskId = new ConnectorTaskId(CONNECTOR_IDS.get(0), 0);
        assertEquals(1, configState.taskCount(CONNECTOR_IDS.get(0)));
        assertEquals(SAMPLE_CONFIGS.get(1), configState.taskConfig(taskId));

        verify(configUpdateListener).onConnectorConfigUpdate(eq(CONNECTOR_IDS.get(0)));
        verify(configUpdateListener).onTaskConfigUpdate(eq(Collections.singleton(taskId)));
    }