private void processConnectorConfigUpdates(Set<String> connectorConfigUpdates) {
        // If we only have connector config updates, we can just bounce the updated connectors that are
        // currently assigned to this worker.
        Set<String> localConnectors = assignment == null ? Collections.emptySet() : new HashSet<>(assignment.connectors());
        Collection<Callable<Void>> connectorsToStart = new ArrayList<>();
        log.trace("Processing connector config updates; "
                + "currently-owned connectors are {}, and to-be-updated connectors are {}",
                localConnectors,
                connectorConfigUpdates);
        for (String connectorName : connectorConfigUpdates) {
            if (!localConnectors.contains(connectorName)) {
                log.trace("Skipping config update for connector {} as it is not owned by this worker",
                        connectorName);
                continue;
            }
            boolean remains = configState.contains(connectorName);
            log.info("Handling connector-only config update by {} connector {}",
                    remains ? "restarting" : "stopping", connectorName);
            try (TickThreadStage stage = new TickThreadStage("stopping connector " + connectorName)) {
                worker.stopAndAwaitConnector(connectorName);
            }
            // The update may be a deletion, so verify we actually need to restart the connector
            if (remains) {
                connectorsToStart.add(getConnectorStartingCallable(connectorName));
            }
        }
        String stageDescription = "restarting " + connectorsToStart.size() + " reconfigured connectors";
        startAndStop(connectorsToStart, stageDescription);
    }
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