@Override
    protected void modifyConnectorOffsets(String connName, Map<Map<String, ?>, Map<String, ?>> offsets, Callback<Message> callback) {
        log.trace("Submitting {} offsets request for connector '{}'", offsets == null ? "reset" : "alter", connName);

        addRequest(() -> {
            if (!modifyConnectorOffsetsChecks(connName, callback)) {
                return null;
            }
            // At this point, we should be the leader (the call to modifyConnectorOffsetsChecks makes sure of that) and can safely run
            // a zombie fencing request
            if (isSourceConnector(connName) && config.exactlyOnceSourceEnabled()) {
                log.debug("Performing a round of zombie fencing before modifying offsets for source connector {} with exactly-once support enabled.", connName);
                doFenceZombieSourceTasks(connName, (error, ignored) -> {
                    if (error != null) {
                        log.error("Failed to perform zombie fencing for source connector prior to modifying offsets", error);
                        callback.onCompletion(new ConnectException("Failed to perform zombie fencing for source connector prior to modifying offsets", error), null);
                    } else {
                        log.debug("Successfully completed zombie fencing for source connector {}; proceeding to modify offsets.", connName);
                        // We need to ensure that we perform the necessary checks again before proceeding to actually altering / resetting the connector offsets since
                        // zombie fencing is done asynchronously and the conditions could have changed since the previous check
                        addRequest(() -> {
                            try (TickThreadStage stage = new TickThreadStage("modifying offsets for connector " + connName)) {
                                if (modifyConnectorOffsetsChecks(connName, callback)) {
                                    worker.modifyConnectorOffsets(connName, configState.connectorConfig(connName), offsets, callback);
                                }
                            }
                            return null;
                        }, forwardErrorAndTickThreadStages(callback));
                    }
                });
            } else {
                try (TickThreadStage stage = new TickThreadStage("modifying offsets for connector " + connName)) {
                    worker.modifyConnectorOffsets(connName, configState.connectorConfig(connName), offsets, callback);
                }
            }
            return null;
        }, forwardErrorAndTickThreadStages(callback));
    }
Builds and executes a restart plan for the connector and its tasks from <code>request</code>.
Execution of a plan involves triggering the stop of eligible connector/tasks and then queuing the start for eligible connector/tasks.

@param request the request to restart connector and tasks
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java