private void processTargetStateRecord(String connectorName, SchemaAndValue value) {
        boolean removed = false;
        boolean stateChanged = true;
        synchronized (lock) {
            if (value.value() == null) {
                // When connector configs are removed, we also write tombstones for the target state.
                log.debug("Removed target state for connector {} due to null value in topic.", connectorName);
                connectorTargetStates.remove(connectorName);
                removed = true;

                // If for some reason we still have configs for the connector, add back the default
                // STARTED state to ensure each connector always has a valid target state.
                if (connectorConfigs.containsKey(connectorName))
                    connectorTargetStates.put(connectorName, TargetState.STARTED);
            } else {
                if (!(value.value() instanceof Map)) {
                    log.error("Ignoring target state for connector '{}' because it is in the wrong format: {}", connectorName, className(value.value()));
                    return;
                }
                @SuppressWarnings("unchecked")
                Map<String, Object> valueMap = (Map<String, Object>) value.value();
                Object targetState = valueMap.get("state.v2");
                if (targetState != null && !(targetState instanceof String)) {
                    log.error("Invalid data for target state for connector '{}': 'state.v2' field should be a String but is {}",
                            connectorName, className(targetState));
                    // We don't return here; it's still possible that there's a value we can use in the older state field
                    targetState = null;
                }
                if (targetState == null) {
                    // This record may have been written by an older worker; fall back on the older state field
                    targetState = valueMap.get("state");
                    if (!(targetState instanceof String)) {
                        log.error("Invalid data for target state for connector '{}': 'state' field should be a String but is {}",
                                connectorName, className(targetState));
                        return;
                    }
                }

                try {
                    TargetState state = TargetState.valueOf((String) targetState);
                    log.debug("Setting target state for connector '{}' to {}", connectorName, targetState);
                    TargetState prevState = connectorTargetStates.put(connectorName, state);
                    stateChanged = !state.equals(prevState);
                } catch (IllegalArgumentException e) {
                    log.error("Invalid target state for connector '{}': {}", connectorName, targetState);
                    return;
                }
            }
        }

        // Note that we do not notify the update listener if the target state has been removed.
        // Instead we depend on the removal callback of the connector config itself to notify the worker.
        // We also don't notify the update listener if the connector doesn't exist yet - a scenario that can
        // occur if an explicit initial target state is specified in the connector creation request.
        if (started && !removed && stateChanged && connectorConfigs.containsKey(connectorName))
            updateListener.onConnectorTargetStateChange(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