private void publishConnectorTaskConfigs(String connName, List<Map<String, String>> taskProps, Callback<Void> cb) {
        List<Map<String, String>> rawTaskProps = reverseTransform(connName, configState, taskProps);
        if (!taskConfigsChanged(configState, connName, rawTaskProps)) {
            return;
        }

        if (isLeader()) {
            writeTaskConfigs(connName, rawTaskProps);
            cb.onCompletion(null, null);
        } else if (restClient == null) {
            throw new NotLeaderException("This worker is not able to communicate with the leader of the cluster, "
                    + "which is required for dynamically-reconfiguring connectors. If running MirrorMaker 2 "
                    + "in dedicated mode, consider enabling inter-worker communication via the "
                    + "'dedicated.mode.enable.internal.rest' property.",
                    leaderUrl()
            );
        } else {
            // We cannot forward the request on the same thread because this reconfiguration can happen as a result of connector
            // addition or removal. If we blocked waiting for the response from leader, we may be kicked out of the worker group.
            forwardRequestExecutor.submit(() -> {
                try {
                    String leaderUrl = leaderUrl();
                    if (Utils.isBlank(leaderUrl)) {
                        cb.onCompletion(new ConnectException("Request to leader to " +
                                "reconfigure connector tasks failed " +
                                "because the URL of the leader's REST interface is empty!"), null);
                        return;
                    }
                    String reconfigUrl = namespacedUrl(leaderUrl)
                            .path("connectors")
                            .path(connName)
                            .path("tasks")
                            .build()
                            .toString();
                    log.trace("Forwarding task configurations for connector {} to leader", connName);
                    String stageDescription = "Forwarding task configurations to the leader at " + leaderUrl;
                    try (TemporaryStage stage = new TemporaryStage(stageDescription, cb, time)) {
                        restClient.httpRequest(reconfigUrl, "POST", null, rawTaskProps, sessionKey, requestSignatureAlgorithm);
                    }
                    cb.onCompletion(null, null);
                } catch (ConnectException e) {
                    log.error("Request to leader to reconfigure connector tasks failed", e);
                    cb.onCompletion(e, null);
                }
            });
        }
    }
Request task configs from the connector and write them to the config storage in case the configs are detected to
have changed. This method retries infinitely with exponential backoff in case of any errors.

@param initialRequestTime the time in milliseconds when the original request was made (i.e. before any retries)
@param connName the name of the connector
@param exponentialBackoff {@link ExponentialBackoff} used to calculate the retry backoff duration
@param attempts the number of retry attempts that have been made
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java