public RestartPlan(RestartRequest request, ConnectorStateInfo restartStateInfo) {
        this.request = Objects.requireNonNull(request, "RestartRequest name may not be null");
        this.stateInfo = Objects.requireNonNull(restartStateInfo, "ConnectorStateInfo name may not be null");
        // Collect the task IDs to stop and restart (may be none)
        this.idsToRestart = Collections.unmodifiableList(
                stateInfo.tasks()
                        .stream()
                        .filter(this::isRestarting)
                        .map(taskState -> new ConnectorTaskId(request.connectorName(), taskState.id()))
                        .collect(Collectors.toList())
        );
    }
@Test
    public void testRestartPlan() {
        ConnectorStateInfo.ConnectorState state = new ConnectorStateInfo.ConnectorState(
                AbstractStatus.State.RESTARTING.name(),
                "foo",
                null
        );
        List<TaskState> tasks = new ArrayList<>();
        tasks.add(new TaskState(1, AbstractStatus.State.RUNNING.name(), "worker1", null));
        tasks.add(new TaskState(2, AbstractStatus.State.PAUSED.name(), "worker1", null));
        tasks.add(new TaskState(3, AbstractStatus.State.RESTARTING.name(), "worker1", null));
        tasks.add(new TaskState(4, AbstractStatus.State.DESTROYED.name(), "worker1", null));
        tasks.add(new TaskState(5, AbstractStatus.State.RUNNING.name(), "worker1", null));
        tasks.add(new TaskState(6, AbstractStatus.State.RUNNING.name(), "worker1", null));
        ConnectorStateInfo connectorStateInfo = new ConnectorStateInfo(CONNECTOR_NAME, state, tasks, ConnectorType.SOURCE);

        RestartRequest restartRequest = new RestartRequest(CONNECTOR_NAME, false, true);
        RestartPlan restartPlan = new RestartPlan(restartRequest, connectorStateInfo);

        assertTrue(restartPlan.shouldRestartConnector());
        assertTrue(restartPlan.shouldRestartTasks());
        assertEquals(1, restartPlan.taskIdsToRestart().size());
        assertEquals(3, restartPlan.taskIdsToRestart().iterator().next().task());
        assertTrue(restartPlan.toString().contains("plan to restart connector"));
    }