private void processTargetStateChanges(Set<String> connectorTargetStateChanges) {
        log.trace("Processing target state updates; "
                + "currently-known connectors are {}, and to-be-updated connectors are {}",
                configState.connectors(), connectorTargetStateChanges);
        for (String connector : connectorTargetStateChanges) {
            TargetState targetState = configState.targetState(connector);
            if (!configState.connectors().contains(connector)) {
                log.debug("Received target state change for unknown connector: {}", connector);
                continue;
            }

            // we must propagate the state change to the worker so that the connector's
            // tasks can transition to the new target state
            worker.setTargetState(connector, targetState, (error, newState) -> {
                if (error != null) {
                    log.error("Failed to transition connector to target state", error);
                    return;
                }
                // additionally, if the worker is running the connector itself, then we need to
                // request reconfiguration to ensure that config changes while paused take effect
                if (newState == TargetState.STARTED) {
                    requestTaskReconfiguration(connector);
                }
            });
        }
    }
Create a herder that will form a Connect cluster with other {@link DistributedHerder} instances (in this or other JVMs)
that have the same group ID.

@param config             the configuration for the worker; may not be null
@param time               the clock to use; may not be null
@param worker             the {@link Worker} instance to use; may not be null
@param kafkaClusterId     the identifier of the Kafka cluster to use for internal topics; may not be null
@param statusBackingStore the backing store for statuses; may not be null
@param configBackingStore the backing store for connector configurations; may not be null
@param restUrl            the URL of this herder's REST API; may not be null, but may be an arbitrary placeholder
                          value if this worker does not expose a REST API
@param restClient         a REST client that can be used to issue requests to other workers in the cluster; may
                          be null if inter-worker communication is not enabled
@param connectorClientConfigOverridePolicy the policy specifying the client configuration properties that may be overridden
                                           in connector configurations; may not be null
@param restNamespace      zero or more path elements to prepend to the paths of forwarded REST requests; may be empty, but not null
@param uponShutdown       any {@link AutoCloseable} objects that should be closed when this herder is {@link #stop() stopped},
                          after all services and resources owned by this herder are stopped
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java