private synchronized boolean updateConfigsWithIncrementalCooperative(AtomicReference<Set<String>> connectorConfigUpdatesCopy,
                                                                         AtomicReference<Set<String>> connectorTargetStateChangesCopy,
                                                                         AtomicReference<Set<ConnectorTaskId>> taskConfigUpdatesCopy) {
        boolean retValue = false;
        // This branch is here to avoid creating a snapshot if not needed
        if (needsReconfigRebalance
                || !connectorConfigUpdates.isEmpty()
                || !connectorTargetStateChanges.isEmpty()
                || !taskConfigUpdates.isEmpty()) {
            log.trace("Handling config updates with incremental cooperative rebalancing");
            // Connector reconfigs only need local updates since there is no coordination between workers required.
            // However, if connectors were added or removed, work needs to be rebalanced since we have more work
            // items to distribute among workers.
            configState = configBackingStore.snapshot();

            if (needsReconfigRebalance) {
                log.debug("Requesting rebalance due to reconfiguration of tasks (needsReconfigRebalance: {})",
                        needsReconfigRebalance);
                member.requestRejoin();
                needsReconfigRebalance = false;
                retValue = true;
            }

            if (!connectorConfigUpdates.isEmpty()) {
                // We can't start/stop while locked since starting connectors can cause task updates that will
                // require writing configs, which in turn make callbacks into this class from another thread that
                // require acquiring a lock. This leads to deadlock. Instead, just copy the info we need and process
                // the updates after unlocking.
                connectorConfigUpdatesCopy.set(connectorConfigUpdates);
                connectorConfigUpdates = new HashSet<>();
            }

            if (!connectorTargetStateChanges.isEmpty()) {
                // Similarly for target state changes which can cause connectors to be restarted
                connectorTargetStateChangesCopy.set(connectorTargetStateChanges);
                connectorTargetStateChanges = new HashSet<>();
            }

            if (!taskConfigUpdates.isEmpty()) {
                // Similarly for task config updates
                taskConfigUpdatesCopy.set(taskConfigUpdates);
                taskConfigUpdates = new HashSet<>();
            }
        } else {
            log.trace("Skipping config updates with incremental cooperative rebalancing " 
                + "since no config rebalance is required " 
                + "and there are no connector config, task config, or target state changes pending");
        }
        return retValue;
    }
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