private void processTaskCountRecord(String connectorName, SchemaAndValue value) {
        if (!(value.value() instanceof Map)) {
            log.error("Ignoring task count record for connector '{}' because it is in the wrong format: {}",  connectorName, className(value.value()));
            return;
        }
        @SuppressWarnings("unchecked")
        int taskCount = intValue(((Map<String, Object>) value.value()).get("task-count"));

        log.debug("Setting task count record for connector '{}' to {}", connectorName, taskCount);
        connectorTaskCountRecords.put(connectorName, taskCount);
        // If a task count record appears after the latest task configs, the connector doesn't need a round of zombie
        // fencing before it can start tasks with the latest configs
        connectorsPendingFencing.remove(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