@Override
    public synchronized void removeTaskConfigs(String connector) {
        ConnectorState state = connectors.get(connector);
        if (state == null)
            throw new IllegalArgumentException("Cannot remove tasks for non-existing connector");

        HashSet<ConnectorTaskId> taskIds = new HashSet<>(state.taskConfigs.keySet());
        state.taskConfigs.clear();
        state.appliedConnConfig = null;

        if (updateListener != null)
            updateListener.onTaskConfigUpdate(taskIds);
    }
@Test
    public void testRemoveTaskConfigs() {
        // Can't remove task configs for non-existent connector
        assertThrows(IllegalArgumentException.class,
            () -> configStore.removeTaskConfigs(CONNECTOR_IDS.get(0)));

        // This workaround is required to verify the arguments passed to ConfigBackingStore.UpdateListener::onTaskConfigUpdate because Mockito
        // records references to the collection instead of a copy and the argument passed to the method the first time in
        // MemoryConfigBackingStore::putTaskConfigs is cleared in MemoryConfigBackingStore::removeTaskConfigs
        final List<Collection<ConnectorTaskId>> onTaskConfigUpdateCaptures = new ArrayList<>();
        doAnswer(invocation -> {
            onTaskConfigUpdateCaptures.add(new HashSet<>(invocation.getArgument(0)));
            return null;
        }).when(configUpdateListener).onTaskConfigUpdate(anySet());

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

        assertEquals(0, configState.taskCount(CONNECTOR_IDS.get(0)));
        assertEquals(Collections.emptyList(), configState.tasks(CONNECTOR_IDS.get(0)));

        verify(configUpdateListener).onConnectorConfigUpdate(eq(CONNECTOR_IDS.get(0)));
        verify(configUpdateListener, times(2)).onTaskConfigUpdate(anySet());

        ConnectorTaskId taskId = new ConnectorTaskId(CONNECTOR_IDS.get(0), 0);
        assertEquals(Arrays.asList(Collections.singleton(taskId), Collections.singleton(taskId)), onTaskConfigUpdateCaptures);
    }