@Override
    public synchronized void restartConnectorAndTasks(RestartRequest request, Callback<ConnectorStateInfo> cb) {
        // Ensure the connector exists
        String connectorName = request.connectorName();
        if (!configState.contains(connectorName)) {
            cb.onCompletion(new NotFoundException("Unknown connector: " + connectorName, null), null);
            return;
        }

        Optional<RestartPlan> maybePlan = buildRestartPlan(request);
        if (!maybePlan.isPresent()) {
            cb.onCompletion(new NotFoundException("Status for connector " + connectorName + " not found", null), null);
            return;
        }
        RestartPlan plan = maybePlan.get();

        // If requested, stop the connector and any tasks, marking each as restarting
        log.info("Received {}", plan);
        if (plan.shouldRestartConnector()) {
            worker.stopAndAwaitConnector(connectorName);
            onRestart(connectorName);
        }
        if (plan.shouldRestartTasks()) {
            // Stop the tasks and mark as restarting
            worker.stopAndAwaitTasks(plan.taskIdsToRestart());
            plan.taskIdsToRestart().forEach(this::onRestart);
        }

        // Now restart the connector and tasks
        if (plan.shouldRestartConnector()) {
            log.debug("Restarting connector '{}'", connectorName);
            startConnector(connectorName, (error, targetState) -> {
                if (error == null) {
                    log.info("Connector '{}' restart successful", connectorName);
                } else {
                    log.error("Connector '{}' restart failed", connectorName, error);
                }
            });
        }
        if (plan.shouldRestartTasks()) {
            log.debug("Restarting {} of {} tasks for {}", plan.restartTaskCount(), plan.totalTaskCount(), request);
            createConnectorTasks(connectorName, plan.taskIdsToRestart());
            log.debug("Restarted {} of {} tasks for {} as requested", plan.restartTaskCount(), plan.totalTaskCount(), request);
        }
        // Complete the restart request
        log.info("Completed {}", plan);
        cb.onCompletion(null, plan.restartConnectorStateInfo());
    }
@Test
    public void testRestartConnectorAndTasksUnknownConnector() {
        initialize(false);
        FutureCallback<ConnectorStateInfo> restartCallback = new FutureCallback<>();
        RestartRequest restartRequest = new RestartRequest("UnknownConnector", false, true);
        herder.restartConnectorAndTasks(restartRequest, restartCallback);
        ExecutionException ee = assertThrows(ExecutionException.class, () -> restartCallback.get(WAIT_TIME_MS, TimeUnit.MILLISECONDS));
        assertInstanceOf(NotFoundException.class, ee.getCause());
    }