@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
    @Timeout(value = 120_000, unit = TimeUnit.MILLISECONDS)
    void testJobSubmissionWithoutUserArtifact() throws Exception {
        try (final TestRestServerEndpoint restServerEndpoint =
                createRestServerEndpoint(new TestJobSubmitHandler())) {
            try (RestClusterClient<?> restClusterClient =
                    createRestClusterClient(restServerEndpoint.getServerAddress().getPort())) {

                restClusterClient.submitJob(jobGraph).get();
            }
        }
    }