protected synchronized void doRestartConnectorAndTasks(RestartRequest request) {
        String connectorName = request.connectorName();
        Optional<RestartPlan> maybePlan = buildRestartPlan(request);
        if (!maybePlan.isPresent()) {
            log.debug("Skipping restart of connector '{}' since no status is available: {}", connectorName, request);
            return;
        }
        RestartPlan plan = maybePlan.get();
        log.info("Executing {}", plan);

        // If requested, stop the connector and any tasks, marking each as restarting
        final ExtendedAssignment currentAssignments = assignment;
        final Collection<ConnectorTaskId> assignedIdsToRestart = plan.taskIdsToRestart()
                .stream()
                .filter(taskId -> currentAssignments.tasks().contains(taskId))
                .collect(Collectors.toList());
        final boolean restartConnector = plan.shouldRestartConnector() && currentAssignments.connectors().contains(connectorName);
        final boolean restartTasks = !assignedIdsToRestart.isEmpty();
        if (restartConnector) {
            String stageDescription = "stopping to-be-restarted connector " + connectorName;
            try (TickThreadStage stage = new TickThreadStage(stageDescription)) {
                worker.stopAndAwaitConnector(connectorName);
            }
            onRestart(connectorName);
        }
        if (restartTasks) {
            String stageDescription = "stopping " + assignedIdsToRestart.size() + " to-be-restarted tasks for connector " + connectorName;
            // Stop the tasks and mark as restarting
            try (TickThreadStage stage = new TickThreadStage(stageDescription)) {
                worker.stopAndAwaitTasks(assignedIdsToRestart);
            }
            assignedIdsToRestart.forEach(this::onRestart);
        }

        // Now restart the connector and tasks
        if (restartConnector) {
            try {
                startConnector(connectorName, (error, targetState) -> {
                    if (error == null) {
                        log.info("Connector '{}' restart successful", connectorName);
                    } else {
                        log.error("Connector '{}' restart failed", connectorName, error);
                    }
                });
            } catch (Throwable t) {
                log.error("Connector '{}' restart failed", connectorName, t);
            }
        }
        if (restartTasks) {
            log.debug("Restarting {} of {} tasks for {}", assignedIdsToRestart.size(), plan.totalTaskCount(), request);
            assignedIdsToRestart.forEach(taskId -> {
                try {
                    if (startTask(taskId)) {
                        log.info("Task '{}' restart successful", taskId);
                    } else {
                        log.error("Task '{}' restart failed", taskId);
                    }
                } catch (Throwable t) {
                    log.error("Task '{}' restart failed", taskId, t);
                }
            });
            log.debug("Restarted {} of {} tasks for {} as requested", assignedIdsToRestart.size(), plan.totalTaskCount(), request);
        }
        log.info("Completed {}", plan);
    }
@Test
    public void testDoRestartConnectorAndTasksNoAssignments() {
        ConnectorTaskId taskId = new ConnectorTaskId(CONN1, 0);
        RestartRequest restartRequest = new RestartRequest(CONN1, false, true);
        RestartPlan restartPlan = mock(RestartPlan.class);
        when(restartPlan.shouldRestartConnector()).thenReturn(true);
        when(restartPlan.taskIdsToRestart()).thenReturn(Collections.singletonList(taskId));

        doReturn(Optional.of(restartPlan)).when(herder).buildRestartPlan(restartRequest);

        herder.assignment = ExtendedAssignment.empty();
        herder.doRestartConnectorAndTasks(restartRequest);

        verifyNoMoreInteractions(restartPlan, worker, member, configBackingStore, statusBackingStore);
    }