private void processTasksCommitRecord(String connectorName, SchemaAndValue value) {
        List<ConnectorTaskId> updatedTasks = new ArrayList<>();
        synchronized (lock) {
            // Edge case: connector was deleted before these task configs were published,
            // but compaction took place and both the original connector config and the
            // tombstone message for it have been removed from the config topic
            // We should ignore these task configs
            Map<String, String> appliedConnectorConfig = connectorConfigs.get(connectorName);
            if (appliedConnectorConfig == null) {
                processConnectorRemoval(connectorName);
                log.debug(
                        "Ignoring task configs for connector {}; it appears that the connector was deleted previously "
                            + "and that log compaction has since removed any trace of its previous configurations "
                            + "from the config topic",
                        connectorName
                );
                return;
            }

            // Apply any outstanding deferred task updates for the given connector. Note that just because we
            // encounter a commit message does not mean it will result in consistent output. In particular due to
            // compaction, there may be cases where . For example if we have the following sequence of writes:
            //
            // 1. Write connector "foo"'s config
            // 2. Write connector "foo", task 1's config <-- compacted
            // 3. Write connector "foo", task 2's config
            // 4. Write connector "foo" task commit message
            // 5. Write connector "foo", task 1's config
            // 6. Write connector "foo", task 2's config
            // 7. Write connector "foo" task commit message
            //
            // then when a new worker starts up, if message 2 had been compacted, then when message 4 is applied
            // "foo" will not have a complete set of configs. Only when message 7 is applied will the complete
            // configuration be available. Worse, if the leader died while writing messages 5, 6, and 7 such that
            // only 5 was written, then there may be nothing that will finish writing the configs and get the
            // log back into a consistent state.
            //
            // It is expected that the user of this class (i.e., the Herder) will take the necessary action to
            // resolve this (i.e., get the connector to recommit its configuration). This inconsistent state is
            // exposed in the snapshots provided via ClusterConfigState so they are easy to handle.
            if (!(value.value() instanceof Map)) { // Schema-less, so we get maps instead of structs
                log.error("Ignoring connector tasks configuration commit for connector '{}' because it is in the wrong format: {}", connectorName, className(value.value()));
                return;
            }
            Map<ConnectorTaskId, Map<String, String>> deferred = deferredTaskUpdates.get(connectorName);

            @SuppressWarnings("unchecked")
            int newTaskCount = intValue(((Map<String, Object>) value.value()).get("tasks"));

            // Validate the configs we're supposed to update to ensure we're getting a complete configuration
            // update of all tasks that are expected based on the number of tasks in the commit message.
            Set<Integer> taskIdSet = taskIds(connectorName, deferred);
            if (!completeTaskIdSet(taskIdSet, newTaskCount)) {
                // Given the logic for writing commit messages, we should only hit this condition due to compacted
                // historical data, in which case we would not have applied any updates yet and there will be no
                // task config data already committed for the connector, so we shouldn't have to clear any data
                // out. All we need to do is add the flag marking it inconsistent.
                log.debug("We have an incomplete set of task configs for connector '{}' probably due to compaction. So we are not doing anything with the new configuration.", connectorName);
                inconsistent.add(connectorName);
            } else {
                if (deferred != null) {
                    taskConfigs.putAll(deferred);
                    updatedTasks.addAll(deferred.keySet());
                    connectorTaskConfigGenerations.compute(connectorName, (ignored, generation) -> generation != null ? generation + 1 : 0);
                }
                inconsistent.remove(connectorName);

                appliedConnectorConfigs.put(
                        connectorName,
                        new AppliedConnectorConfig(appliedConnectorConfig)
                );
            }
            // Always clear the deferred entries, even if we didn't apply them. If they represented an inconsistent
            // update, then we need to see a completely fresh set of configs after this commit message, so we don't
            // want any of these outdated configs
            if (deferred != null)
                deferred.clear();

            connectorTaskCounts.put(connectorName, newTaskCount);
        }

        // If task configs appear after the latest task count record, the connector needs a new round of zombie fencing
        // before it can start tasks with these configs
        connectorsPendingFencing.add(connectorName);
        if (started)
            updateListener.onTaskConfigUpdate(updatedTasks);
    }
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