@Override
    public CompletableFuture<Acknowledge> cancelJob(JobID jobId, Time timeout) {
        jobCancelled = true;
        return super.cancelJob(jobId, timeout);
    }
@Test
    public void testShutdownIfJobCancelledInNormalMode() throws Exception {
        final MiniDispatcher miniDispatcher =
                createMiniDispatcher(ClusterEntrypoint.ExecutionMode.NORMAL);
        miniDispatcher.start();

        try {
            // wait until we have submitted the job
            final TestingJobManagerRunner testingJobManagerRunner =
                    testingJobManagerRunnerFactory.takeCreatedJobManagerRunner();

            assertFalse(miniDispatcher.getTerminationFuture().isDone());

            final DispatcherGateway dispatcherGateway =
                    miniDispatcher.getSelfGateway(DispatcherGateway.class);

            dispatcherGateway.cancelJob(jobGraph.getJobID(), Time.seconds(10L));
            testingJobManagerRunner.completeResultFuture(
                    new ExecutionGraphInfo(
                            new ArchivedExecutionGraphBuilder()
                                    .setJobID(jobGraph.getJobID())
                                    .setState(JobStatus.CANCELED)
                                    .build()));

            ApplicationStatus applicationStatus = miniDispatcher.getShutDownFuture().get();
            assertThat(applicationStatus, is(ApplicationStatus.CANCELED));
        } finally {
            RpcUtils.terminateRpcEndpoint(miniDispatcher);
        }
    }