public Optional<RestartPlan> buildRestartPlan(RestartRequest request) {
        String connectorName = request.connectorName();
        ConnectorStatus connectorStatus = statusBackingStore.get(connectorName);
        if (connectorStatus == null) {
            return Optional.empty();
        }

        // If requested, mark the connector as restarting
        AbstractStatus.State connectorState = request.shouldRestartConnector(connectorStatus) ? AbstractStatus.State.RESTARTING : connectorStatus.state();
        ConnectorStateInfo.ConnectorState connectorInfoState = new ConnectorStateInfo.ConnectorState(
                connectorState.toString(),
                connectorStatus.workerId(),
                connectorStatus.trace()
        );

        // Collect the task states, If requested, mark the task as restarting
        List<ConnectorStateInfo.TaskState> taskStates = statusBackingStore.getAll(connectorName)
                .stream()
                .map(taskStatus -> {
                    AbstractStatus.State taskState = request.shouldRestartTask(taskStatus) ? AbstractStatus.State.RESTARTING : taskStatus.state();
                    return new ConnectorStateInfo.TaskState(
                            taskStatus.id().task(),
                            taskState.toString(),
                            taskStatus.workerId(),
                            taskStatus.trace()
                    );
                })
                .collect(Collectors.toList());
        // Construct the response from the various states
        Map<String, String> conf = rawConfig(connectorName);
        ConnectorStateInfo stateInfo = new ConnectorStateInfo(
                connectorName,
                connectorInfoState,
                taskStates,
                connectorType(conf)
        );
        return Optional.of(new RestartPlan(request, stateInfo));
    }
Build the {@link RestartPlan} that describes what should and should not be restarted given the restart request
and the current status of the connector and task instances.

@param request the restart request; may not be null
@return the restart plan, or empty if this worker has no status for the connector named in the request and therefore the
        connector cannot be restarted
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java