@Override
    public void restartConnectorAndTasks(RestartRequest request, Callback<ConnectorStateInfo> callback) {
        final String connectorName = request.connectorName();
        addRequest(
                () -> {
                    if (checkRebalanceNeeded(callback)) {
                        return null;
                    }
                    if (!configState.connectors().contains(request.connectorName())) {
                        callback.onCompletion(new NotFoundException("Unknown connector: " + connectorName), null);
                        return null;
                    }
                    if (isLeader()) {
                        // Write a restart request to the config backing store, to be executed asynchronously in tick()
                        String stageDescription = "writing restart request for connector " + request.connectorName() + " to the config topic";
                        try (TickThreadStage stage = new TickThreadStage(stageDescription)) {
                            configBackingStore.putRestartRequest(request);
                        }
                        // Compute and send the response that this was accepted
                        Optional<RestartPlan> plan = buildRestartPlan(request);
                        if (!plan.isPresent()) {
                            callback.onCompletion(new NotFoundException("Status for connector " + connectorName + " not found", null), null);
                        } else {
                            callback.onCompletion(null, plan.get().restartConnectorStateInfo());
                        }
                    } else {
                        callback.onCompletion(new NotLeaderException("Only the leader can process restart requests.", leaderUrl()), null);
                    }
                    return null;
                },
                forwardErrorAndTickThreadStages(callback)
        );
    }
Close resources managed by this herder but which are not explicitly started.
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java