@Override
    public synchronized void patchConnectorConfig(String connName, Map<String, String> configPatch, Callback<Created<ConnectorInfo>> callback) {
        try {
            ConnectorInfo connectorInfo = connectorInfo(connName);
            if (connectorInfo == null) {
                callback.onCompletion(new NotFoundException("Connector " + connName + " not found", null), null);
                return;
            }

            Map<String, String> patchedConfig = ConnectUtils.patchConfig(connectorInfo.config(), configPatch);
            validateConnectorConfig(patchedConfig, (error, configInfos) -> {
                if (error != null) {
                    callback.onCompletion(error, null);
                    return;
                }

                requestExecutorService.submit(
                        () -> putConnectorConfig(connName, patchedConfig, null, true, callback, configInfos)
                );
            });
        } catch (Throwable e) {
            callback.onCompletion(e, null);
        }
    }
@Test
    public void testPatchConnectorConfig() throws ExecutionException, InterruptedException, TimeoutException {
        initialize(true);
        // Create the connector.
        Map<String, String> originalConnConfig = connectorConfig(SourceSink.SOURCE);
        originalConnConfig.put("foo0", "unaffected");
        originalConnConfig.put("foo1", "will-be-changed");
        originalConnConfig.put("foo2", "will-be-removed");

        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");

        expectAdd(SourceSink.SOURCE, false, false, false);
        expectConfigValidation(SourceSink.SOURCE, originalConnConfig, patchedConnConfig);

        expectConnectorStartingWithoutTasks(originalConnConfig, true);

        herder.putConnectorConfig(CONNECTOR_NAME, originalConnConfig, false, createCallback);
        createCallback.get(1000L, TimeUnit.SECONDS);

        expectConnectorStartingWithoutTasks(patchedConnConfig, false);

        FutureCallback<Herder.Created<ConnectorInfo>> patchCallback = new FutureCallback<>();
        herder.patchConnectorConfig(CONNECTOR_NAME, connConfigPatch, patchCallback);

        Map<String, String> returnedConfig = patchCallback.get(1000L, TimeUnit.SECONDS).result().config();
        assertEquals(patchedConnConfig, returnedConfig);

        // Also check the returned config when requested.
        FutureCallback<Map<String, String>> configCallback = new FutureCallback<>();
        herder.connectorConfig(CONNECTOR_NAME, configCallback);

        Map<String, String> returnedConfig2 = configCallback.get(1000L, TimeUnit.SECONDS);
        assertEquals(patchedConnConfig, returnedConfig2);
    }