@Override
    public CompletableFuture<JobID> submitJob(@Nonnull JobGraph jobGraph) {
        CompletableFuture<java.nio.file.Path> jobGraphFileFuture =
                CompletableFuture.supplyAsync(
                        () -> {
                            try {
                                final java.nio.file.Path jobGraphFile =
                                        Files.createTempFile(
                                                "flink-jobgraph-" + jobGraph.getJobID(), ".bin");
                                try (ObjectOutputStream objectOut =
                                        new ObjectOutputStream(
                                                Files.newOutputStream(jobGraphFile))) {
                                    objectOut.writeObject(jobGraph);
                                }
                                return jobGraphFile;
                            } catch (IOException e) {
                                throw new CompletionException(
                                        new FlinkException("Failed to serialize JobGraph.", e));
                            }
                        },
                        executorService);

        CompletableFuture<Tuple2<JobSubmitRequestBody, Collection<FileUpload>>> requestFuture =
                jobGraphFileFuture.thenApply(
                        jobGraphFile -> {
                            List<String> jarFileNames = new ArrayList<>(8);
                            List<JobSubmitRequestBody.DistributedCacheFile> artifactFileNames =
                                    new ArrayList<>(8);
                            Collection<FileUpload> filesToUpload = new ArrayList<>(8);

                            filesToUpload.add(
                                    new FileUpload(
                                            jobGraphFile, RestConstants.CONTENT_TYPE_BINARY));

                            for (Path jar : jobGraph.getUserJars()) {
                                jarFileNames.add(jar.getName());
                                filesToUpload.add(
                                        new FileUpload(
                                                Paths.get(jar.toUri()),
                                                RestConstants.CONTENT_TYPE_JAR));
                            }

                            for (Map.Entry<String, DistributedCache.DistributedCacheEntry>
                                    artifacts : jobGraph.getUserArtifacts().entrySet()) {
                                final Path artifactFilePath =
                                        new Path(artifacts.getValue().filePath);
                                try {
                                    // Only local artifacts need to be uploaded.
                                    if (!artifactFilePath.getFileSystem().isDistributedFS()) {
                                        artifactFileNames.add(
                                                new JobSubmitRequestBody.DistributedCacheFile(
                                                        artifacts.getKey(),
                                                        artifactFilePath.getName()));
                                        filesToUpload.add(
                                                new FileUpload(
                                                        Paths.get(artifactFilePath.getPath()),
                                                        RestConstants.CONTENT_TYPE_BINARY));
                                    }
                                } catch (IOException e) {
                                    throw new CompletionException(
                                            new FlinkException(
                                                    "Failed to get the FileSystem of artifact "
                                                            + artifactFilePath
                                                            + ".",
                                                    e));
                                }
                            }

                            final JobSubmitRequestBody requestBody =
                                    new JobSubmitRequestBody(
                                            jobGraphFile.getFileName().toString(),
                                            jarFileNames,
                                            artifactFileNames);

                            return Tuple2.of(
                                    requestBody, Collections.unmodifiableCollection(filesToUpload));
                        });

        final CompletableFuture<JobSubmitResponseBody> submissionFuture =
                requestFuture.thenCompose(
                        requestAndFileUploads -> {
                            LOG.info(
                                    "Submitting job '{}' ({}).",
                                    jobGraph.getName(),
                                    jobGraph.getJobID());
                            return sendRetriableRequest(
                                    JobSubmitHeaders.getInstance(),
                                    EmptyMessageParameters.getInstance(),
                                    requestAndFileUploads.f0,
                                    requestAndFileUploads.f1,
                                    isConnectionProblemOrServiceUnavailable(),
                                    (receiver, error) -> {
                                        if (error != null) {
                                            LOG.warn(
                                                    "Attempt to submit job '{}' ({}) to '{}' has failed.",
                                                    jobGraph.getName(),
                                                    jobGraph.getJobID(),
                                                    receiver,
                                                    error);
                                        } else {
                                            LOG.info(
                                                    "Successfully submitted job '{}' ({}) to '{}'.",
                                                    jobGraph.getName(),
                                                    jobGraph.getJobID(),
                                                    receiver);
                                        }
                                    });
                        });

        submissionFuture
                .exceptionally(ignored -> null) // ignore errors
                .thenCompose(ignored -> jobGraphFileFuture)
                .thenAccept(
                        jobGraphFile -> {
                            try {
                                Files.delete(jobGraphFile);
                            } catch (IOException e) {
                                LOG.warn("Could not delete temporary file {}.", jobGraphFile, e);
                            }
                        });

        return submissionFuture
                .thenApply(ignore -> jobGraph.getJobID())
                .exceptionally(
                        (Throwable throwable) -> {
                            throw new CompletionException(
                                    new JobSubmissionException(
                                            jobGraph.getJobID(),
                                            "Failed to submit JobGraph.",
                                            ExceptionUtils.stripCompletionException(throwable)));
                        });
    }
@Test
    void testSubmitJobAndWaitForExecutionResult() throws Exception {
        final TestJobExecutionResultHandler testJobExecutionResultHandler =
                new TestJobExecutionResultHandler(
                        new RestHandlerException(
                                "should trigger retry", HttpResponseStatus.SERVICE_UNAVAILABLE),
                        JobExecutionResultResponseBody.inProgress(),
                        // On an UNKNOWN JobResult it should be retried
                        JobExecutionResultResponseBody.created(
                                new JobResult.Builder()
                                        .applicationStatus(ApplicationStatus.UNKNOWN)
                                        .jobId(jobId)
                                        .netRuntime(Long.MAX_VALUE)
                                        .accumulatorResults(
                                                Collections.singletonMap(
                                                        "testName",
                                                        new SerializedValue<>(
                                                                OptionalFailure.of(1.0))))
                                        .build()),
                        JobExecutionResultResponseBody.created(
                                new JobResult.Builder()
                                        .applicationStatus(ApplicationStatus.SUCCEEDED)
                                        .jobId(jobId)
                                        .netRuntime(Long.MAX_VALUE)
                                        .accumulatorResults(
                                                Collections.singletonMap(
                                                        "testName",
                                                        new SerializedValue<>(
                                                                OptionalFailure.of(1.0))))
                                        .build()),
                        JobExecutionResultResponseBody.created(
                                new JobResult.Builder()
                                        .applicationStatus(ApplicationStatus.FAILED)
                                        .jobId(jobId)
                                        .netRuntime(Long.MAX_VALUE)
                                        .serializedThrowable(
                                                new SerializedThrowable(
                                                        new RuntimeException("expected")))
                                        .build()));

        // Fail the first JobExecutionResult HTTP polling attempt, which should not be a problem
        // because of the retries.
        final AtomicBoolean firstExecutionResultPollFailed = new AtomicBoolean(false);
        // Fail the first JobSubmit HTTP request, which should not be a problem because of the
        // retries.
        final AtomicBoolean firstSubmitRequestFailed = new AtomicBoolean(false);
        failHttpRequest =
                (messageHeaders, messageParameters, requestBody) -> {
                    messageHeaders =
                            ((UrlPrefixDecorator)
                                            ((CustomHeadersDecorator) messageHeaders)
                                                    .getDecorated())
                                    .getDecorated();
                    if (messageHeaders instanceof JobExecutionResultHeaders) {
                        return !firstExecutionResultPollFailed.getAndSet(true);
                    }
                    if (messageHeaders instanceof JobSubmitHeaders) {
                        return !firstSubmitRequestFailed.getAndSet(true);
                    }
                    return false;
                };

        try (TestRestServerEndpoint restServerEndpoint =
                createRestServerEndpoint(
                        testJobExecutionResultHandler, new TestJobSubmitHandler())) {

            try (RestClusterClient<?> restClusterClient =
                    createRestClusterClient(restServerEndpoint.getServerAddress().getPort())) {
                final JobExecutionResult jobExecutionResult =
                        restClusterClient
                                .submitJob(jobGraph)
                                .thenCompose(restClusterClient::requestJobResult)
                                .get()
                                .toJobExecutionResult(ClassLoader.getSystemClassLoader());
                assertThat(firstExecutionResultPollFailed).isTrue();
                assertThat(firstSubmitRequestFailed).isTrue();
                assertThat(jobExecutionResult.getJobID()).isEqualTo(jobId);
                assertThat(jobExecutionResult.getNetRuntime()).isEqualTo(Long.MAX_VALUE);
                assertThat(jobExecutionResult.getAllAccumulatorResults())
                        .hasSize(1)
                        .containsEntry("testName", 1.0);

                try {
                    restClusterClient
                            .submitJob(jobGraph)
                            .thenCompose(restClusterClient::requestJobResult)
                            .get()
                            .toJobExecutionResult(ClassLoader.getSystemClassLoader());
                    fail("Expected exception not thrown.");
                } catch (final Exception e) {
                    final Optional<RuntimeException> cause =
                            ExceptionUtils.findThrowable(e, RuntimeException.class);
                    assertThat(cause).isPresent();
                    assertThat(cause.get().getMessage()).isEqualTo("expected");
                }
            }
        }
    }