private void processConnectorConfigRecord(String connectorName, SchemaAndValue value) {
        boolean removed = false;
        synchronized (lock) {
            if (value.value() == null) {
                // Connector deletion will be written as a null value
                processConnectorRemoval(connectorName);
                log.info("Successfully processed removal of connector '{}'", connectorName);
                removed = true;
            } else {
                // Connector configs can be applied and callbacks invoked immediately
                if (!(value.value() instanceof Map)) {
                    log.error("Ignoring configuration for connector '{}' because it is in the wrong format: {}", connectorName, className(value.value()));
                    return;
                }
                @SuppressWarnings("unchecked")
                Object newConnectorConfig = ((Map<String, Object>) value.value()).get("properties");
                if (!(newConnectorConfig instanceof Map)) {
                    log.error("Invalid data for config for connector '{}': 'properties' field should be a Map but is {}",
                            connectorName, className(newConnectorConfig));
                    return;
                }
                log.debug("Updating configuration for connector '{}'", connectorName);
                @SuppressWarnings("unchecked")
                Map<String, String> stringsConnectorConfig = (Map<String, String>) newConnectorConfig;
                connectorConfigs.put(connectorName, stringsConnectorConfig);

                // Set the initial state of the connector to STARTED, which ensures that any connectors
                // which were created with 0.9 Connect will be initialized in the STARTED state.
                if (!connectorTargetStates.containsKey(connectorName))
                    connectorTargetStates.put(connectorName, TargetState.STARTED);
            }
        }
        if (started) {
            if (removed)
                updateListener.onConnectorConfigRemove(connectorName);
            else
                updateListener.onConnectorConfigUpdate(connectorName);
        }
    }
Send one or more records to the config topic synchronously. Note that {@link #claimWritePrivileges()} must be
successfully invoked before calling this method if this store is configured to use a fencable writer.
@param keyValues the list of producer record key/value pairs
@param timer Timer bounding how long this method can block. The timer is updated before the method returns.
connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java