@Override
    public synchronized void putTargetState(String connector, TargetState state) {
        ConnectorState connectorState = connectors.get(connector);
        if (connectorState == null)
            throw new IllegalArgumentException("No connector `" + connector + "` configured");

        TargetState prevState = connectorState.targetState;
        connectorState.targetState = state;

        if (updateListener != null && !state.equals(prevState))
            updateListener.onConnectorTargetStateChange(connector);
    }
@Test
    public void testPutTargetState() {
        // Can't write target state for non-existent connector
        assertThrows(IllegalArgumentException.class, () -> configStore.putTargetState(CONNECTOR_IDS.get(0), TargetState.PAUSED));

        configStore.putConnectorConfig(CONNECTOR_IDS.get(0), SAMPLE_CONFIGS.get(0), null);
        configStore.putTargetState(CONNECTOR_IDS.get(0), TargetState.PAUSED);
        // Ensure that ConfigBackingStore.UpdateListener::onConnectorTargetStateChange is called only once if the same state is written twice
        configStore.putTargetState(CONNECTOR_IDS.get(0), TargetState.PAUSED);
        ClusterConfigState configState = configStore.snapshot();

        assertEquals(TargetState.PAUSED, configState.targetState(CONNECTOR_IDS.get(0)));

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