private void writeToConfigTopicAsLeader(String description, Runnable write) {
        try (TickThreadStage stage = new TickThreadStage(description)) {
            write.run();
        } catch (PrivilegedWriteException e) {
            log.warn("Failed to write to config topic as leader; will rejoin group if necessary and, if still leader, attempt to reclaim write privileges for the config topic", e);
            fencedFromConfigTopic = true;
            throw new ConnectException("Failed to write to config topic; this may be due to a transient error and the request can be safely retried", e);
        }
    }
Perform an action that writes to the config topic, and if it fails because the leader has been fenced out, make note of that
fact so that we can try to reclaim write ownership (if still the leader of the cluster) in a subsequent iteration of the tick loop.
Note that it is not necessary to wrap every write to the config topic in this method, only the writes that should be performed
exclusively by the leader. For example, {@link ConfigBackingStore#putTargetState(String, TargetState)} does not require this
method, as it can be invoked by any worker in the cluster.
@param description a description of the action; e.g., "writing 5 task configs for connector file-source"
@param write the action that writes to the config topic, such as {@link ConfigBackingStore#putSessionKey(SessionKey)} or
             {@link ConfigBackingStore#putConnectorConfig(String, Map, TargetState)}.
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java