private boolean checkForKeyRotation(long now) {
        SecretKey key;
        long expiration;
        synchronized (this) {
            // This happens on startup; the snapshot contains the session key,
            // but no callback in the config update listener has been fired for it yet.
            if (sessionKey == null && configState.sessionKey() != null) {
                sessionKey = configState.sessionKey().key();
                keyExpiration = configState.sessionKey().creationTimestamp() + keyRotationIntervalMs;
            }
            key = sessionKey;
            expiration = keyExpiration;
        }

        if (internalRequestValidationEnabled()) {
            if (isLeader()) {
                if (key == null) {
                    log.debug("Internal request signing is enabled but no session key has been distributed yet. "
                        + "Distributing new key now.");
                    return true;
                } else if (expiration <= now) {
                    log.debug("Existing key has expired. Distributing new key now.");
                    return true;
                } else if (!key.getAlgorithm().equals(keyGenerator.getAlgorithm())
                        || key.getEncoded().length != keyGenerator.generateKey().getEncoded().length) {
                    log.debug("Previously-distributed key uses different algorithm/key size "
                        + "than required by current worker configuration. Distributing new key now.");
                    return true;
                }
            }
        }
        return false;
    }
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