@Override
    public void patchConnectorConfig(String connName, Map<String, String> configPatch, Callback<Created<ConnectorInfo>> callback) {
        log.trace("Submitting connector config patch request {}", connName);
        addRequest(() -> {
            // This reduces (but not completely eliminates) the chance for race conditions.
            if (!isLeader()) {
                callback.onCompletion(new NotLeaderException("Only the leader can set connector configs.", leaderUrl()), null);
                return null;
            }

            ConnectorInfo connectorInfo = connectorInfo(connName);
            if (connectorInfo == null) {
                callback.onCompletion(new NotFoundException("Connector " + connName + " not found", null), null);
            } else {
                Map<String, String> patchedConfig = ConnectUtils.patchConfig(connectorInfo.config(), configPatch);
                doPutConnectorConfig(connName, patchedConfig, null, true, callback);
            }
            return null;
        }, forwardErrorAndTickThreadStages(callback));
    }
@Test
    public void testPatchConnectorConfig() throws Exception {
        when(herder.connectorType(anyMap())).thenReturn(ConnectorType.SOURCE);
        when(member.memberId()).thenReturn("leader");
        when(statusBackingStore.connectors()).thenReturn(Collections.emptySet());

        Map<String, String> originalConnConfig = new HashMap<>(CONN1_CONFIG);
        originalConnConfig.put("foo0", "unaffected");
        originalConnConfig.put("foo1", "will-be-changed");
        originalConnConfig.put("foo2", "will-be-removed");

        // The connector is pre-existing due to the mocks.

        ClusterConfigState originalSnapshot = new ClusterConfigState(
                1,
                null,
                Collections.singletonMap(CONN1, 0),
                Collections.singletonMap(CONN1, originalConnConfig),
                Collections.singletonMap(CONN1, TargetState.STARTED),
                Collections.emptyMap(),
                Collections.emptyMap(),
                Collections.emptyMap(),
                Collections.emptyMap(),
                Collections.emptySet(),
                Collections.emptySet());
        expectConfigRefreshAndSnapshot(originalSnapshot);
        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);

        expectMemberPoll();

        // Patch the connector config.
        Map<String, String> connConfigPatch = new HashMap<>();
        connConfigPatch.put("foo1", "changed");
        connConfigPatch.put("foo2", null);
        connConfigPatch.put("foo3", "added");

        Map<String, String> patchedConnConfig = new HashMap<>(originalConnConfig);
        patchedConnConfig.put("foo0", "unaffected");
        patchedConnConfig.put("foo1", "changed");
        patchedConnConfig.remove("foo2");
        patchedConnConfig.put("foo3", "added");

        expectRebalance(1, singletonList(CONN1), Collections.emptyList(), true);

        ArgumentCaptor<Callback<ConfigInfos>> validateCallback = ArgumentCaptor.forClass(Callback.class);
        doAnswer(invocation -> {
            validateCallback.getValue().onCompletion(null, CONN1_CONFIG_INFOS);
            return null;
        }).when(herder).validateConnectorConfig(eq(patchedConnConfig), validateCallback.capture());

        FutureCallback<Herder.Created<ConnectorInfo>> patchCallback = new FutureCallback<>();
        herder.patchConnectorConfig(CONN1, connConfigPatch, patchCallback);
        herder.tick();
        assertTrue(patchCallback.isDone());
        assertEquals(patchedConnConfig, patchCallback.get().result().config());

        // This is effectively the main check of this test:
        // we validate that what's written in the config storage is the patched config.
        verify(configBackingStore).putConnectorConfig(eq(CONN1), eq(patchedConnConfig), isNull());
        verifyNoMoreInteractions(configBackingStore);

        // No need to check herder.connectorConfig explicitly:
        // all the related parts are mocked and that the config is correct is checked by eq()'s in the mocked called above.
    }