Github user zentol commented on a diff in the pull request: https://github.com/apache/flink/pull/6203#discussion_r198178787 --- Diff: flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java --- @@ -315,42 +315,58 @@ public JobSubmissionResult submitJob(JobGraph jobGraph, ClassLoader classLoader) // we have to enable queued scheduling because slot will be allocated lazily jobGraph.setAllowQueuedScheduling(true); - log.info("Requesting blob server port."); - CompletableFuture<BlobServerPortResponseBody> portFuture = sendRequest(BlobServerPortHeaders.getInstance()); + CompletableFuture<JobSubmitResponseBody> submissionFuture = CompletableFuture.supplyAsync( + () -> { + log.info("Submitting job graph."); + + List<String> jarFileNames = new ArrayList<>(8); + List<JobSubmitRequestBody.DistributedCacheFile> artifactFileNames = new ArrayList<>(8); + Collection<FileUpload> filesToUpload = new ArrayList<>(8); - CompletableFuture<JobGraph> jobUploadFuture = portFuture.thenCombine( - getDispatcherAddress(), - (BlobServerPortResponseBody response, String dispatcherAddress) -> { - final int blobServerPort = response.port; - final InetSocketAddress address = new InetSocketAddress(dispatcherAddress, blobServerPort); - final List<PermanentBlobKey> keys; + // TODO: need configurable location + final java.nio.file.Path jobGraphFile; try { - log.info("Uploading jar files."); - keys = BlobClient.uploadFiles(address, flinkConfig, jobGraph.getJobID(), jobGraph.getUserJars()); - jobGraph.uploadUserArtifacts(address, flinkConfig); - } catch (IOException ioe) { - throw new CompletionException(new FlinkException("Could not upload job files.", ioe)); + jobGraphFile = Files.createTempFile("flink-jobgraph", ".bin"); + try (OutputStream fileOut = Files.newOutputStream(jobGraphFile)) { + try (ObjectOutputStream objectOut = new ObjectOutputStream(fileOut)) { + objectOut.writeObject(jobGraph); + } + } + filesToUpload.add(new FileUpload(jobGraphFile, RestConstants.CONTENT_TYPE_BINARY)); + } catch (IOException e) { + throw new RuntimeException("lol", e); --- End diff -- needs a proper exception
---