void processRestartRequests() {
        List<RestartRequest> restartRequests;
        synchronized (this) {
            if (pendingRestartRequests.isEmpty()) {
                return;
            }
            //dequeue into a local list to minimize the work being done within the synchronized block
            restartRequests = new ArrayList<>(pendingRestartRequests.values());
            pendingRestartRequests.clear();
        }
        restartRequests.forEach(restartRequest -> {
            String stageDescription = "handling restart request for connector " + restartRequest.connectorName();
            try (TickThreadStage stage = new TickThreadStage(stageDescription)) {
                doRestartConnectorAndTasks(restartRequest);
            } catch (Exception e) {
                log.warn("Unexpected error while trying to process " + restartRequest + ", the restart request will be skipped.", e);
            }
        });
    }
Process all pending restart requests. There can be at most one request per connector.

<p>This method is called from within the {@link #tick()} method.
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java