private void processTaskConfigRecord(ConnectorTaskId taskId, SchemaAndValue value) {
        synchronized (lock) {
            if (value.value() == null) {
                log.error("Ignoring task configuration for task {} because it is unexpectedly null", taskId);
                return;
            }
            if (!(value.value() instanceof Map)) {
                log.error("Ignoring task configuration for task {} because the value is not a Map but is {}", taskId, className(value.value()));
                return;
            }

            @SuppressWarnings("unchecked")
            Object newTaskConfig = ((Map<String, Object>) value.value()).get("properties");
            if (!(newTaskConfig instanceof Map)) {
                log.error("Invalid data for config of task {} 'properties' field should be a Map but is {}", taskId, className(newTaskConfig));
                return;
            }

            Map<ConnectorTaskId, Map<String, String>> deferred = deferredTaskUpdates.computeIfAbsent(taskId.connector(), k -> new HashMap<>());
            log.debug("Storing new config for task {}; this will wait for a commit message before the new config will take effect.", taskId);
            @SuppressWarnings("unchecked")
            Map<String, String> stringsTaskConfig = (Map<String, String>) newTaskConfig;
            deferred.put(taskId, stringsTaskConfig);
        }
    }
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