You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by zentol <gi...@git.apache.org> on 2018/06/21 21:13:56 UTC
[GitHub] flink pull request #6199: [FLINK-9624][runtime] Move jar/artifact upload out...
GitHub user zentol opened a pull request:
https://github.com/apache/flink/pull/6199
[FLINK-9624][runtime] Move jar/artifact upload out of jobgraph
## What is the purpose of the change
This PR moves the logic for uploading jars/artifacts from the jobgraph into a separate utility class usable by all submission methods.
The new `ClientUtils` class exposes 2 methods for uploading jars/artifacts and setting the respective blob keys on the `JobGraph`.
All existing job-submission method were updated to use the new utilities and should now behave the same.
The subsumed methods in `JobGraph` were removed, but remnants of them remain in 2 added methods:
* setUserArtifactBlobKey sets the blobkey for a specific entry
* finalizeUserArtifactEntries writes the artifact entries into the `ExecutionConfig`
## Verifying this change
* `ClientUtils` is tested in `ClientUtilsTest`
* `JobGraph` changes are covered in `JobGraphTest`
* client modifications are covered by various existing tests
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/zentol/flink 9280_delta
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/6199.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #6199
----
commit 13e3dc7dc9c0b7205223368a460993a309cb58ad
Author: zentol <ch...@...>
Date: 2018-06-13T16:21:21Z
[FLINK-9624][runtime] Move jar/artifact upload out of jobgraph
----
---
[GitHub] flink pull request #6199: [FLINK-9624][runtime] Move jar/artifact upload out...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/6199#discussion_r197496529
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java ---
@@ -552,96 +550,33 @@ public boolean hasUsercodeJarFiles() {
return this.userJarBlobKeys;
}
- /**
- * Uploads the previously added user JAR files to the job manager through
- * the job manager's BLOB server. The BLOB servers' address is given as a
- * parameter. This function issues a blocking call.
- *
- * @param blobServerAddress of the blob server to upload the jars to
- * @param blobClientConfig the blob client configuration
- * @throws IOException Thrown, if the file upload to the JobManager failed.
- */
- public void uploadUserJars(
- InetSocketAddress blobServerAddress,
- Configuration blobClientConfig) throws IOException {
- if (!userJars.isEmpty()) {
- List<PermanentBlobKey> blobKeys = BlobClient.uploadFiles(
- blobServerAddress, blobClientConfig, jobID, userJars);
-
- for (PermanentBlobKey blobKey : blobKeys) {
- if (!userJarBlobKeys.contains(blobKey)) {
- userJarBlobKeys.add(blobKey);
- }
- }
- }
- }
-
@Override
public String toString() {
return "JobGraph(jobId: " + jobID + ")";
}
- /**
- * Configures JobGraph with user specified artifacts. If the files are in local system it uploads them
- * to the BLOB server, otherwise it just puts metadata for future remote access from within task executor.
- *
- * @param blobServerAddress of the blob server to upload the files to
- * @param blobClientConfig the blob client configuration
- * @throws IOException Thrown, if the file upload to the Blob server failed.
- */
- public void uploadUserArtifacts(
- InetSocketAddress blobServerAddress,
- Configuration blobClientConfig) throws IOException {
-
- Set<Map.Entry<String, DistributedCache.DistributedCacheEntry>> uploadToBlobServer = new HashSet<>();
- Set<Map.Entry<String, DistributedCache.DistributedCacheEntry>> distributeViaDFS = new HashSet<>();
-
- for (Map.Entry<String, DistributedCache.DistributedCacheEntry> userArtifact : userArtifacts.entrySet()) {
- Path filePath = new Path(userArtifact.getValue().filePath);
-
- try {
- if (filePath.getFileSystem().isDistributedFS()) {
- distributeViaDFS.add(userArtifact);
- } else {
- uploadToBlobServer.add(userArtifact);
- }
-
- } catch (IOException ex) {
- distributeViaDFS.add(userArtifact);
- }
+ public void setUserArtifactBlobKey(String entryName, PermanentBlobKey blobKey) {
+ byte[] serializedBlobKey;
+ try {
+ serializedBlobKey = InstantiationUtil.serializeObject(blobKey);
+ } catch (IOException e) {
+ throw new FlinkRuntimeException("Could not serialize blobkey " + blobKey + ".", e);
}
+ userArtifacts.computeIfPresent(entryName, (key, originalEntry) -> new DistributedCache.DistributedCacheEntry(
+ originalEntry.filePath,
+ originalEntry.isExecutable,
+ serializedBlobKey,
+ originalEntry.isZipped
+ ));
+ }
- uploadViaBlob(blobServerAddress, blobClientConfig, uploadToBlobServer);
-
- for (Map.Entry<String, DistributedCache.DistributedCacheEntry> userArtifact : distributeViaDFS) {
+ public void finalizeUserArtifactEntries() {
--- End diff --
https://issues.apache.org/jira/browse/FLINK-8713
---
[GitHub] flink pull request #6199: [FLINK-9624][runtime] Move jar/artifact upload out...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/6199#discussion_r197437750
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java ---
@@ -552,96 +550,33 @@ public boolean hasUsercodeJarFiles() {
return this.userJarBlobKeys;
}
- /**
- * Uploads the previously added user JAR files to the job manager through
- * the job manager's BLOB server. The BLOB servers' address is given as a
- * parameter. This function issues a blocking call.
- *
- * @param blobServerAddress of the blob server to upload the jars to
- * @param blobClientConfig the blob client configuration
- * @throws IOException Thrown, if the file upload to the JobManager failed.
- */
- public void uploadUserJars(
- InetSocketAddress blobServerAddress,
- Configuration blobClientConfig) throws IOException {
- if (!userJars.isEmpty()) {
- List<PermanentBlobKey> blobKeys = BlobClient.uploadFiles(
- blobServerAddress, blobClientConfig, jobID, userJars);
-
- for (PermanentBlobKey blobKey : blobKeys) {
- if (!userJarBlobKeys.contains(blobKey)) {
- userJarBlobKeys.add(blobKey);
- }
- }
- }
- }
-
@Override
public String toString() {
return "JobGraph(jobId: " + jobID + ")";
}
- /**
- * Configures JobGraph with user specified artifacts. If the files are in local system it uploads them
- * to the BLOB server, otherwise it just puts metadata for future remote access from within task executor.
- *
- * @param blobServerAddress of the blob server to upload the files to
- * @param blobClientConfig the blob client configuration
- * @throws IOException Thrown, if the file upload to the Blob server failed.
- */
- public void uploadUserArtifacts(
- InetSocketAddress blobServerAddress,
- Configuration blobClientConfig) throws IOException {
-
- Set<Map.Entry<String, DistributedCache.DistributedCacheEntry>> uploadToBlobServer = new HashSet<>();
- Set<Map.Entry<String, DistributedCache.DistributedCacheEntry>> distributeViaDFS = new HashSet<>();
-
- for (Map.Entry<String, DistributedCache.DistributedCacheEntry> userArtifact : userArtifacts.entrySet()) {
- Path filePath = new Path(userArtifact.getValue().filePath);
-
- try {
- if (filePath.getFileSystem().isDistributedFS()) {
- distributeViaDFS.add(userArtifact);
- } else {
- uploadToBlobServer.add(userArtifact);
- }
-
- } catch (IOException ex) {
- distributeViaDFS.add(userArtifact);
- }
+ public void setUserArtifactBlobKey(String entryName, PermanentBlobKey blobKey) {
+ byte[] serializedBlobKey;
+ try {
+ serializedBlobKey = InstantiationUtil.serializeObject(blobKey);
+ } catch (IOException e) {
+ throw new FlinkRuntimeException("Could not serialize blobkey " + blobKey + ".", e);
}
+ userArtifacts.computeIfPresent(entryName, (key, originalEntry) -> new DistributedCache.DistributedCacheEntry(
+ originalEntry.filePath,
+ originalEntry.isExecutable,
+ serializedBlobKey,
+ originalEntry.isZipped
+ ));
+ }
- uploadViaBlob(blobServerAddress, blobClientConfig, uploadToBlobServer);
-
- for (Map.Entry<String, DistributedCache.DistributedCacheEntry> userArtifact : distributeViaDFS) {
+ public void finalizeUserArtifactEntries() {
--- End diff --
I think we would not need this method if we don't write the `DistributedCacheEntries` into the configuration. If I'm not mistaken, then we send the `userArtifacts` map anyway to the cluster. The things which are missing are: Addind a serial version UID to the `DistributedCacheEntry`, and adding the `userArtifacts` to the `TaskDeploymentDescriptor` to send them to the `TaskManager`.
---
[GitHub] flink pull request #6199: [FLINK-9624][runtime] Move jar/artifact upload out...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/6199#discussion_r197438267
--- Diff: flink-runtime/src/test/java/org/apache/flink/runtime/client/ClientUtilsTest.java ---
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.client;
+
+import org.apache.flink.api.common.cache.DistributedCache;
+import org.apache.flink.configuration.BlobServerOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.blob.BlobClient;
+import org.apache.flink.runtime.blob.BlobServer;
+import org.apache.flink.runtime.blob.VoidBlobStore;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.file.Files;
+import java.util.Arrays;
+import java.util.Collection;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * TODO: add javadoc.
+ */
+public class ClientUtilsTest extends TestLogger {
+
+ @ClassRule
+ public static TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ private static BlobServer blobServer;
+
+ @BeforeClass
+ public static void setup() throws IOException {
+ Configuration config = new Configuration();
+ config.setString(BlobServerOptions.STORAGE_DIRECTORY,
+ temporaryFolder.newFolder().getAbsolutePath());
+ blobServer = new BlobServer(config, new VoidBlobStore());
+ blobServer.start();
+ }
+
+ @Test
+ public void uploadAndSetUserJars() throws IOException {
+ java.nio.file.Path tmpDir = temporaryFolder.newFolder().toPath();
+ JobGraph jobGraph = new JobGraph();
+
+ Collection<Path> jars = Arrays.asList(
+ new Path(Files.createFile(tmpDir.resolve("jar1.jar")).toString()),
+ new Path(Files.createFile(tmpDir.resolve("jar2.jar")).toString()));
+
+ jars.forEach(jobGraph::addJar);
+
+ assertEquals(jars.size(), jobGraph.getUserJars().size());
+ assertEquals(0, jobGraph.getUserJarBlobKeys().size());
+
+ try (BlobClient blobClient = new BlobClient(new InetSocketAddress("localhost", blobServer.getPort()), new Configuration())) {
+ ClientUtils.uploadAndSetUserJars(jobGraph, blobClient);
+ }
+
+ assertEquals(jars.size(), jobGraph.getUserJars().size());
+ assertEquals(jars.size(), jobGraph.getUserJarBlobKeys().size());
+ assertEquals(jars.size(), jobGraph.getUserJarBlobKeys().stream().distinct().count());
--- End diff --
Assert that we find the blob keys in the blob upload directory.
---
[GitHub] flink pull request #6199: [FLINK-9624][runtime] Move jar/artifact upload out...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/6199#discussion_r197279258
--- Diff: flink-runtime/src/test/java/org/apache/flink/runtime/client/ClientUtilsTest.java ---
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.client;
+
+import org.apache.flink.api.common.cache.DistributedCache;
+import org.apache.flink.configuration.BlobServerOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.blob.BlobClient;
+import org.apache.flink.runtime.blob.BlobServer;
+import org.apache.flink.runtime.blob.VoidBlobStore;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.file.Files;
+import java.util.Arrays;
+import java.util.Collection;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * TODO: add javadoc.
--- End diff --
missing javadoc
---
[GitHub] flink pull request #6199: [FLINK-9624][runtime] Move jar/artifact upload out...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/6199#discussion_r197434489
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java ---
@@ -552,96 +550,33 @@ public boolean hasUsercodeJarFiles() {
return this.userJarBlobKeys;
}
- /**
- * Uploads the previously added user JAR files to the job manager through
- * the job manager's BLOB server. The BLOB servers' address is given as a
- * parameter. This function issues a blocking call.
- *
- * @param blobServerAddress of the blob server to upload the jars to
- * @param blobClientConfig the blob client configuration
- * @throws IOException Thrown, if the file upload to the JobManager failed.
- */
- public void uploadUserJars(
- InetSocketAddress blobServerAddress,
- Configuration blobClientConfig) throws IOException {
- if (!userJars.isEmpty()) {
- List<PermanentBlobKey> blobKeys = BlobClient.uploadFiles(
- blobServerAddress, blobClientConfig, jobID, userJars);
-
- for (PermanentBlobKey blobKey : blobKeys) {
- if (!userJarBlobKeys.contains(blobKey)) {
- userJarBlobKeys.add(blobKey);
- }
- }
- }
- }
-
@Override
public String toString() {
return "JobGraph(jobId: " + jobID + ")";
}
- /**
- * Configures JobGraph with user specified artifacts. If the files are in local system it uploads them
- * to the BLOB server, otherwise it just puts metadata for future remote access from within task executor.
- *
- * @param blobServerAddress of the blob server to upload the files to
- * @param blobClientConfig the blob client configuration
- * @throws IOException Thrown, if the file upload to the Blob server failed.
- */
- public void uploadUserArtifacts(
- InetSocketAddress blobServerAddress,
- Configuration blobClientConfig) throws IOException {
-
- Set<Map.Entry<String, DistributedCache.DistributedCacheEntry>> uploadToBlobServer = new HashSet<>();
- Set<Map.Entry<String, DistributedCache.DistributedCacheEntry>> distributeViaDFS = new HashSet<>();
-
- for (Map.Entry<String, DistributedCache.DistributedCacheEntry> userArtifact : userArtifacts.entrySet()) {
- Path filePath = new Path(userArtifact.getValue().filePath);
-
- try {
- if (filePath.getFileSystem().isDistributedFS()) {
- distributeViaDFS.add(userArtifact);
- } else {
- uploadToBlobServer.add(userArtifact);
- }
-
- } catch (IOException ex) {
- distributeViaDFS.add(userArtifact);
- }
+ public void setUserArtifactBlobKey(String entryName, PermanentBlobKey blobKey) {
+ byte[] serializedBlobKey;
+ try {
+ serializedBlobKey = InstantiationUtil.serializeObject(blobKey);
+ } catch (IOException e) {
+ throw new FlinkRuntimeException("Could not serialize blobkey " + blobKey + ".", e);
--- End diff --
I would not throw a `FlinkRuntimeException` here. Instead we could led the `IOException` bubble up.
---
[GitHub] flink pull request #6199: [FLINK-9624][runtime] Move jar/artifact upload out...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/6199#discussion_r197279550
--- Diff: flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java ---
@@ -323,17 +325,18 @@ public JobSubmissionResult submitJob(JobGraph jobGraph, ClassLoader classLoader)
(BlobServerPortResponseBody response, String dispatcherAddress) -> {
final int blobServerPort = response.port;
final InetSocketAddress address = new InetSocketAddress(dispatcherAddress, blobServerPort);
- final List<PermanentBlobKey> keys;
- 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));
- }
- for (PermanentBlobKey key : keys) {
- jobGraph.addUserJarBlobKey(key);
+ List<Path> userJars = jobGraph.getUserJars();
+ Map<String, DistributedCache.DistributedCacheEntry> userArtifacts = jobGraph.getUserArtifacts();
--- End diff --
this entire block is effectively duplicated in several classes and could also be moved to `ClientUtils`, but I wasn't sure whether this wouldn't put too much logic into a single method,
---
[GitHub] flink issue #6199: [FLINK-9624][runtime] Move jar/artifact upload out of job...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on the issue:
https://github.com/apache/flink/pull/6199
Test failure seems to be unrelated.
---
[GitHub] flink pull request #6199: [FLINK-9624][runtime] Move jar/artifact upload out...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/6199#discussion_r197430670
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/client/ClientUtils.java ---
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.client;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.cache.DistributedCache;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.blob.BlobClient;
+import org.apache.flink.runtime.blob.PermanentBlobKey;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Map;
+
+/**
+ * Contains utility methods for clients.
+ */
+public enum ClientUtils {
+ ;
+
+ /**
+ * Uploads the user jars from the given {@link JobGraph} using the given {@link BlobClient},
+ * and sets the appropriate blobkeys.
+ *
+ * @param jobGraph jobgraph requiring user jars
+ * @param blobClient client to upload jars with
+ * @throws IOException if the upload fails
+ */
+ public static void uploadAndSetUserJars(JobGraph jobGraph, BlobClient blobClient) throws IOException {
+ Collection<PermanentBlobKey> blobKeys = uploadUserJars(jobGraph.getJobID(), jobGraph.getUserJars(), blobClient);
+ setUserJarBlobKeys(blobKeys, jobGraph);
+ }
+
+ private static Collection<PermanentBlobKey> uploadUserJars(JobID jobId, Collection<Path> userJars, BlobClient blobClient) throws IOException {
+ Collection<PermanentBlobKey> blobKeys = new ArrayList<>(userJars.size());
+ for (Path jar : userJars) {
+ final PermanentBlobKey blobKey = blobClient.uploadFile(jobId, jar);
+ blobKeys.add(blobKey);
+ }
+ return blobKeys;
+ }
+
+ private static void setUserJarBlobKeys(Collection<PermanentBlobKey> blobKeys, JobGraph jobGraph) {
+ blobKeys.forEach(jobGraph::addUserJarBlobKey);
+ }
+
+ /**
+ * Uploads the user artifacts from the given {@link JobGraph} using the given {@link BlobClient},
+ * and sets the appropriate blobkeys.
+ *
+ * @param jobGraph jobgraph requiring user artifacts
+ * @param blobClient client to upload artifacts with
+ * @throws IOException if the upload fails
+ */
+ public static void uploadAndSetUserArtifacts(JobGraph jobGraph, BlobClient blobClient) throws IOException {
+ Collection<Tuple2<String, PermanentBlobKey>> blobKeys = uploadUserArtifacts(jobGraph.getJobID(), jobGraph.getUserArtifacts(), blobClient);
+ setUserArtifactBlobKeys(jobGraph, blobKeys);
+ }
+
+ private static Collection<Tuple2<String, PermanentBlobKey>> uploadUserArtifacts(JobID jobID, Map<String, DistributedCache.DistributedCacheEntry> userArtifacts, BlobClient blobClient) throws IOException {
+ Collection<Tuple2<String, PermanentBlobKey>> blobKeys = new ArrayList<>(userArtifacts.size());
+ for (Map.Entry<String, DistributedCache.DistributedCacheEntry> userArtifact : userArtifacts.entrySet()) {
+ Path path = new Path(userArtifact.getValue().filePath);
+ // only upload local files
+ if (!path.getFileSystem().isDistributedFS()) {
+ final PermanentBlobKey blobKey = blobClient.uploadFile(jobID, new Path(userArtifact.getValue().filePath));
--- End diff --
we could reuse `path` here
---
[GitHub] flink pull request #6199: [FLINK-9624][runtime] Move jar/artifact upload out...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/6199#discussion_r197279905
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/client/ClientUtils.java ---
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.client;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.cache.DistributedCache;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.blob.BlobClient;
+import org.apache.flink.runtime.blob.PermanentBlobKey;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Map;
+
+/**
+ * Contains utility methods for clients.
+ */
+public enum ClientUtils {
+ ;
+
+ /**
+ * Uploads the user jars from the given {@link JobGraph} using the given {@link BlobClient},
+ * and sets the appropriate blobkeys.
+ *
+ * @param jobGraph jobgraph requiring user jars
+ * @param blobClient client to upload jars with
+ * @throws IOException if the upload fails
+ */
+ public static void uploadAndSetUserJars(JobGraph jobGraph, BlobClient blobClient) throws IOException {
--- End diff --
`JarRunHandler` could use this method as well.
---
[GitHub] flink pull request #6199: [FLINK-9624][runtime] Move jar/artifact upload out...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/6199#discussion_r197438415
--- Diff: flink-runtime/src/test/java/org/apache/flink/runtime/client/ClientUtilsTest.java ---
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.client;
+
+import org.apache.flink.api.common.cache.DistributedCache;
+import org.apache.flink.configuration.BlobServerOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.blob.BlobClient;
+import org.apache.flink.runtime.blob.BlobServer;
+import org.apache.flink.runtime.blob.VoidBlobStore;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.file.Files;
+import java.util.Arrays;
+import java.util.Collection;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * TODO: add javadoc.
+ */
+public class ClientUtilsTest extends TestLogger {
+
+ @ClassRule
+ public static TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ private static BlobServer blobServer;
+
+ @BeforeClass
+ public static void setup() throws IOException {
+ Configuration config = new Configuration();
+ config.setString(BlobServerOptions.STORAGE_DIRECTORY,
+ temporaryFolder.newFolder().getAbsolutePath());
+ blobServer = new BlobServer(config, new VoidBlobStore());
+ blobServer.start();
+ }
+
+ @Test
+ public void uploadAndSetUserJars() throws IOException {
+ java.nio.file.Path tmpDir = temporaryFolder.newFolder().toPath();
+ JobGraph jobGraph = new JobGraph();
+
+ Collection<Path> jars = Arrays.asList(
+ new Path(Files.createFile(tmpDir.resolve("jar1.jar")).toString()),
+ new Path(Files.createFile(tmpDir.resolve("jar2.jar")).toString()));
+
+ jars.forEach(jobGraph::addJar);
+
+ assertEquals(jars.size(), jobGraph.getUserJars().size());
+ assertEquals(0, jobGraph.getUserJarBlobKeys().size());
+
+ try (BlobClient blobClient = new BlobClient(new InetSocketAddress("localhost", blobServer.getPort()), new Configuration())) {
+ ClientUtils.uploadAndSetUserJars(jobGraph, blobClient);
+ }
+
+ assertEquals(jars.size(), jobGraph.getUserJars().size());
+ assertEquals(jars.size(), jobGraph.getUserJarBlobKeys().size());
+ assertEquals(jars.size(), jobGraph.getUserJarBlobKeys().stream().distinct().count());
+ }
+
+ @Test
+ public void uploadAndSetUserArtifacts() throws IOException {
+ java.nio.file.Path tmpDir = temporaryFolder.newFolder().toPath();
+ JobGraph jobGraph = new JobGraph();
+
+ Collection<DistributedCache.DistributedCacheEntry> localArtifacts = Arrays.asList(
+ new DistributedCache.DistributedCacheEntry(Files.createFile(tmpDir.resolve("art1")).toString(), true, true),
+ new DistributedCache.DistributedCacheEntry(Files.createFile(tmpDir.resolve("art2")).toString(), true, false),
+ new DistributedCache.DistributedCacheEntry(Files.createFile(tmpDir.resolve("art3")).toString(), false, true),
+ new DistributedCache.DistributedCacheEntry(Files.createFile(tmpDir.resolve("art4")).toString(), true, false)
+ );
+
+ Collection<DistributedCache.DistributedCacheEntry> distributedArtifacts = Arrays.asList(
+ new DistributedCache.DistributedCacheEntry("hdfs://localhost:1234/test", true, false)
+ );
+
+ for (DistributedCache.DistributedCacheEntry entry : localArtifacts) {
+ jobGraph.addUserArtifact(entry.filePath, entry);
+ }
+ for (DistributedCache.DistributedCacheEntry entry : distributedArtifacts) {
+ jobGraph.addUserArtifact(entry.filePath, entry);
+ }
+
+ final int totalNumArtifacts = localArtifacts.size() + distributedArtifacts.size();
+
+ assertEquals(totalNumArtifacts, jobGraph.getUserArtifacts().size());
+ assertEquals(0, jobGraph.getUserArtifacts().values().stream().filter(entry -> entry.blobKey != null).count());
+
+ try (BlobClient blobClient = new BlobClient(new InetSocketAddress("localhost", blobServer.getPort()), new Configuration())) {
+ ClientUtils.uploadAndSetUserArtifacts(jobGraph, blobClient);
+ }
+
+ assertEquals(totalNumArtifacts, jobGraph.getUserArtifacts().size());
+ assertEquals(localArtifacts.size(), jobGraph.getUserArtifacts().values().stream().filter(entry -> entry.blobKey != null).count());
+ assertEquals(distributedArtifacts.size(), jobGraph.getUserArtifacts().values().stream().filter(entry -> entry.blobKey == null).count());
+ // 1 unique key for each local artifact, and null for distributed artifacts
+ assertEquals(localArtifacts.size() + 1, jobGraph.getUserArtifacts().values().stream().map(entry -> entry.blobKey).distinct().count());
+ for (DistributedCache.DistributedCacheEntry original : localArtifacts) {
+ assertState(original, jobGraph.getUserArtifacts().get(original.filePath), false);
--- End diff --
Assert that the blobs can be found in the blob server storage directory.
---
[GitHub] flink pull request #6199: [FLINK-9624][runtime] Move jar/artifact upload out...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/6199#discussion_r197433655
--- Diff: flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java ---
@@ -323,17 +325,18 @@ public JobSubmissionResult submitJob(JobGraph jobGraph, ClassLoader classLoader)
(BlobServerPortResponseBody response, String dispatcherAddress) -> {
final int blobServerPort = response.port;
final InetSocketAddress address = new InetSocketAddress(dispatcherAddress, blobServerPort);
- final List<PermanentBlobKey> keys;
- 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));
- }
- for (PermanentBlobKey key : keys) {
- jobGraph.addUserJarBlobKey(key);
+ List<Path> userJars = jobGraph.getUserJars();
+ Map<String, DistributedCache.DistributedCacheEntry> userArtifacts = jobGraph.getUserArtifacts();
+ if (!userJars.isEmpty() || !userArtifacts.isEmpty()) {
+ try (BlobClient client = new BlobClient(address, flinkConfig)) {
--- End diff --
I would be in favour of having a `ClientUtils#uploadJobGraphFiles(jobGraph, flinkConfig, Supplier<BlobClient>)` which basically does what's being done here.
---
[GitHub] flink pull request #6199: [FLINK-9624][runtime] Move jar/artifact upload out...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/6199#discussion_r197433761
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/client/ClientUtils.java ---
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.client;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.cache.DistributedCache;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.blob.BlobClient;
+import org.apache.flink.runtime.blob.PermanentBlobKey;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Map;
+
+/**
+ * Contains utility methods for clients.
+ */
+public enum ClientUtils {
+ ;
+
+ /**
+ * Uploads the user jars from the given {@link JobGraph} using the given {@link BlobClient},
+ * and sets the appropriate blobkeys.
+ *
+ * @param jobGraph jobgraph requiring user jars
+ * @param blobClient client to upload jars with
+ * @throws IOException if the upload fails
+ */
+ public static void uploadAndSetUserJars(JobGraph jobGraph, BlobClient blobClient) throws IOException {
--- End diff --
Good point, let's do it then.
---
[GitHub] flink pull request #6199: [FLINK-9624][runtime] Move jar/artifact upload out...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/6199#discussion_r198129013
--- Diff: flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarRunHandler.java ---
@@ -114,17 +114,12 @@ public JarRunHandler(
CompletableFuture<JobGraph> jarUploadFuture = jobGraphFuture.thenCombine(blobServerPortFuture, (jobGraph, blobServerPort) -> {
final InetSocketAddress address = new InetSocketAddress(getDispatcherHost(gateway), blobServerPort);
- final List<PermanentBlobKey> keys;
- try {
- keys = BlobClient.uploadFiles(address, configuration, jobGraph.getJobID(), jobGraph.getUserJars());
+ try (BlobClient blobClient = new BlobClient(address, configuration)) {
+ ClientUtils.uploadAndSetUserJars(jobGraph, blobClient);
--- End diff --
The `JarRunHandler` now also uses `uploadJobGraphFiles()`.
---
[GitHub] flink pull request #6199: [FLINK-9624][runtime] Move jar/artifact upload out...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/6199#discussion_r197487498
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java ---
@@ -552,96 +550,33 @@ public boolean hasUsercodeJarFiles() {
return this.userJarBlobKeys;
}
- /**
- * Uploads the previously added user JAR files to the job manager through
- * the job manager's BLOB server. The BLOB servers' address is given as a
- * parameter. This function issues a blocking call.
- *
- * @param blobServerAddress of the blob server to upload the jars to
- * @param blobClientConfig the blob client configuration
- * @throws IOException Thrown, if the file upload to the JobManager failed.
- */
- public void uploadUserJars(
- InetSocketAddress blobServerAddress,
- Configuration blobClientConfig) throws IOException {
- if (!userJars.isEmpty()) {
- List<PermanentBlobKey> blobKeys = BlobClient.uploadFiles(
- blobServerAddress, blobClientConfig, jobID, userJars);
-
- for (PermanentBlobKey blobKey : blobKeys) {
- if (!userJarBlobKeys.contains(blobKey)) {
- userJarBlobKeys.add(blobKey);
- }
- }
- }
- }
-
@Override
public String toString() {
return "JobGraph(jobId: " + jobID + ")";
}
- /**
- * Configures JobGraph with user specified artifacts. If the files are in local system it uploads them
- * to the BLOB server, otherwise it just puts metadata for future remote access from within task executor.
- *
- * @param blobServerAddress of the blob server to upload the files to
- * @param blobClientConfig the blob client configuration
- * @throws IOException Thrown, if the file upload to the Blob server failed.
- */
- public void uploadUserArtifacts(
- InetSocketAddress blobServerAddress,
- Configuration blobClientConfig) throws IOException {
-
- Set<Map.Entry<String, DistributedCache.DistributedCacheEntry>> uploadToBlobServer = new HashSet<>();
- Set<Map.Entry<String, DistributedCache.DistributedCacheEntry>> distributeViaDFS = new HashSet<>();
-
- for (Map.Entry<String, DistributedCache.DistributedCacheEntry> userArtifact : userArtifacts.entrySet()) {
- Path filePath = new Path(userArtifact.getValue().filePath);
-
- try {
- if (filePath.getFileSystem().isDistributedFS()) {
- distributeViaDFS.add(userArtifact);
- } else {
- uploadToBlobServer.add(userArtifact);
- }
-
- } catch (IOException ex) {
- distributeViaDFS.add(userArtifact);
- }
+ public void setUserArtifactBlobKey(String entryName, PermanentBlobKey blobKey) {
+ byte[] serializedBlobKey;
+ try {
+ serializedBlobKey = InstantiationUtil.serializeObject(blobKey);
+ } catch (IOException e) {
+ throw new FlinkRuntimeException("Could not serialize blobkey " + blobKey + ".", e);
}
+ userArtifacts.computeIfPresent(entryName, (key, originalEntry) -> new DistributedCache.DistributedCacheEntry(
+ originalEntry.filePath,
+ originalEntry.isExecutable,
+ serializedBlobKey,
+ originalEntry.isZipped
+ ));
+ }
- uploadViaBlob(blobServerAddress, blobClientConfig, uploadToBlobServer);
-
- for (Map.Entry<String, DistributedCache.DistributedCacheEntry> userArtifact : distributeViaDFS) {
+ public void finalizeUserArtifactEntries() {
--- End diff --
Alright, please create a follow up JIRA issue.
---
[GitHub] flink pull request #6199: [FLINK-9624][runtime] Move jar/artifact upload out...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/6199#discussion_r197472732
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java ---
@@ -552,96 +550,33 @@ public boolean hasUsercodeJarFiles() {
return this.userJarBlobKeys;
}
- /**
- * Uploads the previously added user JAR files to the job manager through
- * the job manager's BLOB server. The BLOB servers' address is given as a
- * parameter. This function issues a blocking call.
- *
- * @param blobServerAddress of the blob server to upload the jars to
- * @param blobClientConfig the blob client configuration
- * @throws IOException Thrown, if the file upload to the JobManager failed.
- */
- public void uploadUserJars(
- InetSocketAddress blobServerAddress,
- Configuration blobClientConfig) throws IOException {
- if (!userJars.isEmpty()) {
- List<PermanentBlobKey> blobKeys = BlobClient.uploadFiles(
- blobServerAddress, blobClientConfig, jobID, userJars);
-
- for (PermanentBlobKey blobKey : blobKeys) {
- if (!userJarBlobKeys.contains(blobKey)) {
- userJarBlobKeys.add(blobKey);
- }
- }
- }
- }
-
@Override
public String toString() {
return "JobGraph(jobId: " + jobID + ")";
}
- /**
- * Configures JobGraph with user specified artifacts. If the files are in local system it uploads them
- * to the BLOB server, otherwise it just puts metadata for future remote access from within task executor.
- *
- * @param blobServerAddress of the blob server to upload the files to
- * @param blobClientConfig the blob client configuration
- * @throws IOException Thrown, if the file upload to the Blob server failed.
- */
- public void uploadUserArtifacts(
- InetSocketAddress blobServerAddress,
- Configuration blobClientConfig) throws IOException {
-
- Set<Map.Entry<String, DistributedCache.DistributedCacheEntry>> uploadToBlobServer = new HashSet<>();
- Set<Map.Entry<String, DistributedCache.DistributedCacheEntry>> distributeViaDFS = new HashSet<>();
-
- for (Map.Entry<String, DistributedCache.DistributedCacheEntry> userArtifact : userArtifacts.entrySet()) {
- Path filePath = new Path(userArtifact.getValue().filePath);
-
- try {
- if (filePath.getFileSystem().isDistributedFS()) {
- distributeViaDFS.add(userArtifact);
- } else {
- uploadToBlobServer.add(userArtifact);
- }
-
- } catch (IOException ex) {
- distributeViaDFS.add(userArtifact);
- }
+ public void setUserArtifactBlobKey(String entryName, PermanentBlobKey blobKey) {
+ byte[] serializedBlobKey;
+ try {
+ serializedBlobKey = InstantiationUtil.serializeObject(blobKey);
+ } catch (IOException e) {
+ throw new FlinkRuntimeException("Could not serialize blobkey " + blobKey + ".", e);
}
+ userArtifacts.computeIfPresent(entryName, (key, originalEntry) -> new DistributedCache.DistributedCacheEntry(
+ originalEntry.filePath,
+ originalEntry.isExecutable,
+ serializedBlobKey,
+ originalEntry.isZipped
+ ));
+ }
- uploadViaBlob(blobServerAddress, blobClientConfig, uploadToBlobServer);
-
- for (Map.Entry<String, DistributedCache.DistributedCacheEntry> userArtifact : distributeViaDFS) {
+ public void finalizeUserArtifactEntries() {
--- End diff --
I agree that this would nice, but I think that this is out of scope of this PR as we would have to touch an entirely new set of classes.
---
[GitHub] flink issue #6199: [FLINK-9624][runtime] Move jar/artifact upload out of job...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the issue:
https://github.com/apache/flink/pull/6199
merging.
---
[GitHub] flink pull request #6199: [FLINK-9624][runtime] Move jar/artifact upload out...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/6199#discussion_r197435093
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java ---
@@ -552,96 +550,33 @@ public boolean hasUsercodeJarFiles() {
return this.userJarBlobKeys;
}
- /**
- * Uploads the previously added user JAR files to the job manager through
- * the job manager's BLOB server. The BLOB servers' address is given as a
- * parameter. This function issues a blocking call.
- *
- * @param blobServerAddress of the blob server to upload the jars to
- * @param blobClientConfig the blob client configuration
- * @throws IOException Thrown, if the file upload to the JobManager failed.
- */
- public void uploadUserJars(
- InetSocketAddress blobServerAddress,
- Configuration blobClientConfig) throws IOException {
- if (!userJars.isEmpty()) {
- List<PermanentBlobKey> blobKeys = BlobClient.uploadFiles(
- blobServerAddress, blobClientConfig, jobID, userJars);
-
- for (PermanentBlobKey blobKey : blobKeys) {
- if (!userJarBlobKeys.contains(blobKey)) {
- userJarBlobKeys.add(blobKey);
- }
- }
- }
- }
-
@Override
public String toString() {
return "JobGraph(jobId: " + jobID + ")";
}
- /**
- * Configures JobGraph with user specified artifacts. If the files are in local system it uploads them
- * to the BLOB server, otherwise it just puts metadata for future remote access from within task executor.
- *
- * @param blobServerAddress of the blob server to upload the files to
- * @param blobClientConfig the blob client configuration
- * @throws IOException Thrown, if the file upload to the Blob server failed.
- */
- public void uploadUserArtifacts(
- InetSocketAddress blobServerAddress,
- Configuration blobClientConfig) throws IOException {
-
- Set<Map.Entry<String, DistributedCache.DistributedCacheEntry>> uploadToBlobServer = new HashSet<>();
- Set<Map.Entry<String, DistributedCache.DistributedCacheEntry>> distributeViaDFS = new HashSet<>();
-
- for (Map.Entry<String, DistributedCache.DistributedCacheEntry> userArtifact : userArtifacts.entrySet()) {
- Path filePath = new Path(userArtifact.getValue().filePath);
-
- try {
- if (filePath.getFileSystem().isDistributedFS()) {
- distributeViaDFS.add(userArtifact);
- } else {
- uploadToBlobServer.add(userArtifact);
- }
-
- } catch (IOException ex) {
- distributeViaDFS.add(userArtifact);
- }
+ public void setUserArtifactBlobKey(String entryName, PermanentBlobKey blobKey) {
+ byte[] serializedBlobKey;
+ try {
+ serializedBlobKey = InstantiationUtil.serializeObject(blobKey);
+ } catch (IOException e) {
+ throw new FlinkRuntimeException("Could not serialize blobkey " + blobKey + ".", e);
}
+ userArtifacts.computeIfPresent(entryName, (key, originalEntry) -> new DistributedCache.DistributedCacheEntry(
+ originalEntry.filePath,
+ originalEntry.isExecutable,
+ serializedBlobKey,
+ originalEntry.isZipped
+ ));
+ }
- uploadViaBlob(blobServerAddress, blobClientConfig, uploadToBlobServer);
-
- for (Map.Entry<String, DistributedCache.DistributedCacheEntry> userArtifact : distributeViaDFS) {
+ public void finalizeUserArtifactEntries() {
--- End diff --
Maybe rename to `writeUserArtifactEntriesToConfiguration`
---
[GitHub] flink pull request #6199: [FLINK-9624][runtime] Move jar/artifact upload out...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/6199#discussion_r197721680
--- Diff: flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarRunHandler.java ---
@@ -114,17 +114,12 @@ public JarRunHandler(
CompletableFuture<JobGraph> jarUploadFuture = jobGraphFuture.thenCombine(blobServerPortFuture, (jobGraph, blobServerPort) -> {
final InetSocketAddress address = new InetSocketAddress(getDispatcherHost(gateway), blobServerPort);
- final List<PermanentBlobKey> keys;
- try {
- keys = BlobClient.uploadFiles(address, configuration, jobGraph.getJobID(), jobGraph.getUserJars());
+ try (BlobClient blobClient = new BlobClient(address, configuration)) {
+ ClientUtils.uploadAndSetUserJars(jobGraph, blobClient);
--- End diff --
we _could_ use `uploadJobGraphFiles` here, but there isn't really a use-case for uploading distributed cache artifacts when going through the JarRunHandler, since we're already on the server here.
---
[GitHub] flink pull request #6199: [FLINK-9624][runtime] Move jar/artifact upload out...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/6199#discussion_r197377877
--- Diff: flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java ---
@@ -323,17 +325,18 @@ public JobSubmissionResult submitJob(JobGraph jobGraph, ClassLoader classLoader)
(BlobServerPortResponseBody response, String dispatcherAddress) -> {
final int blobServerPort = response.port;
final InetSocketAddress address = new InetSocketAddress(dispatcherAddress, blobServerPort);
- final List<PermanentBlobKey> keys;
- 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));
- }
- for (PermanentBlobKey key : keys) {
- jobGraph.addUserJarBlobKey(key);
+ List<Path> userJars = jobGraph.getUserJars();
+ Map<String, DistributedCache.DistributedCacheEntry> userArtifacts = jobGraph.getUserArtifacts();
+ if (!userJars.isEmpty() || !userArtifacts.isEmpty()) {
+ try (BlobClient client = new BlobClient(address, flinkConfig)) {
--- End diff --
alternatively we could refactor the try-with-resource statement and exception handling into a method that accepts a function, which would be used like this:
```
ClientUtils.withBlobClient(address, flinkConfig, () -> {
log.info("Uploading jar files.");
ClientUtils.uploadAndSetUserJars(jobGraph, client);
log.info("Uploading jar artifacts.");
ClientUtils.uploadAndSetUserArtifacts(jobGraph, client);
}
```
---
[GitHub] flink pull request #6199: [FLINK-9624][runtime] Move jar/artifact upload out...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/6199#discussion_r198030793
--- Diff: flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarRunHandler.java ---
@@ -114,17 +114,12 @@ public JarRunHandler(
CompletableFuture<JobGraph> jarUploadFuture = jobGraphFuture.thenCombine(blobServerPortFuture, (jobGraph, blobServerPort) -> {
final InetSocketAddress address = new InetSocketAddress(getDispatcherHost(gateway), blobServerPort);
- final List<PermanentBlobKey> keys;
- try {
- keys = BlobClient.uploadFiles(address, configuration, jobGraph.getJobID(), jobGraph.getUserJars());
+ try (BlobClient blobClient = new BlobClient(address, configuration)) {
+ ClientUtils.uploadAndSetUserJars(jobGraph, blobClient);
--- End diff --
But it would reduce code redundancy, right? If this is the case, then let's do it.
---
[GitHub] flink pull request #6199: [FLINK-9624][runtime] Move jar/artifact upload out...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/6199#discussion_r197279307
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java ---
@@ -552,96 +550,33 @@ public boolean hasUsercodeJarFiles() {
return this.userJarBlobKeys;
}
- /**
- * Uploads the previously added user JAR files to the job manager through
- * the job manager's BLOB server. The BLOB servers' address is given as a
- * parameter. This function issues a blocking call.
- *
- * @param blobServerAddress of the blob server to upload the jars to
- * @param blobClientConfig the blob client configuration
- * @throws IOException Thrown, if the file upload to the JobManager failed.
- */
- public void uploadUserJars(
- InetSocketAddress blobServerAddress,
- Configuration blobClientConfig) throws IOException {
- if (!userJars.isEmpty()) {
- List<PermanentBlobKey> blobKeys = BlobClient.uploadFiles(
- blobServerAddress, blobClientConfig, jobID, userJars);
-
- for (PermanentBlobKey blobKey : blobKeys) {
- if (!userJarBlobKeys.contains(blobKey)) {
- userJarBlobKeys.add(blobKey);
- }
- }
- }
- }
-
@Override
public String toString() {
return "JobGraph(jobId: " + jobID + ")";
}
- /**
- * Configures JobGraph with user specified artifacts. If the files are in local system it uploads them
- * to the BLOB server, otherwise it just puts metadata for future remote access from within task executor.
- *
- * @param blobServerAddress of the blob server to upload the files to
- * @param blobClientConfig the blob client configuration
- * @throws IOException Thrown, if the file upload to the Blob server failed.
- */
- public void uploadUserArtifacts(
- InetSocketAddress blobServerAddress,
- Configuration blobClientConfig) throws IOException {
-
- Set<Map.Entry<String, DistributedCache.DistributedCacheEntry>> uploadToBlobServer = new HashSet<>();
- Set<Map.Entry<String, DistributedCache.DistributedCacheEntry>> distributeViaDFS = new HashSet<>();
-
- for (Map.Entry<String, DistributedCache.DistributedCacheEntry> userArtifact : userArtifacts.entrySet()) {
- Path filePath = new Path(userArtifact.getValue().filePath);
-
- try {
- if (filePath.getFileSystem().isDistributedFS()) {
- distributeViaDFS.add(userArtifact);
- } else {
- uploadToBlobServer.add(userArtifact);
- }
-
- } catch (IOException ex) {
- distributeViaDFS.add(userArtifact);
- }
+ public void setUserArtifactBlobKey(String entryName, PermanentBlobKey blobKey) {
+ byte[] serializedBlobKey;
+ try {
+ serializedBlobKey = InstantiationUtil.serializeObject(blobKey);
+ } catch (IOException e) {
+ throw new FlinkRuntimeException("Could not serialize blobkey " + blobKey + ".", e);
}
+ userArtifacts.computeIfPresent(entryName, (key, originalEntry) -> new DistributedCache.DistributedCacheEntry(
+ originalEntry.filePath,
+ originalEntry.isExecutable,
+ serializedBlobKey,
+ originalEntry.isZipped
+ ));
+ }
- uploadViaBlob(blobServerAddress, blobClientConfig, uploadToBlobServer);
-
- for (Map.Entry<String, DistributedCache.DistributedCacheEntry> userArtifact : distributeViaDFS) {
+ public void finalizeUserArtifactEntries() {
--- End diff --
missing test
---
[GitHub] flink pull request #6199: [FLINK-9624][runtime] Move jar/artifact upload out...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/6199#discussion_r197467939
--- Diff: flink-runtime/src/test/java/org/apache/flink/runtime/client/ClientUtilsTest.java ---
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.client;
+
+import org.apache.flink.api.common.cache.DistributedCache;
+import org.apache.flink.configuration.BlobServerOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.blob.BlobClient;
+import org.apache.flink.runtime.blob.BlobServer;
+import org.apache.flink.runtime.blob.VoidBlobStore;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.file.Files;
+import java.util.Arrays;
+import java.util.Collection;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * TODO: add javadoc.
+ */
+public class ClientUtilsTest extends TestLogger {
+
+ @ClassRule
+ public static TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ private static BlobServer blobServer;
+
+ @BeforeClass
+ public static void setup() throws IOException {
+ Configuration config = new Configuration();
+ config.setString(BlobServerOptions.STORAGE_DIRECTORY,
+ temporaryFolder.newFolder().getAbsolutePath());
+ blobServer = new BlobServer(config, new VoidBlobStore());
+ blobServer.start();
+ }
+
+ @Test
+ public void uploadAndSetUserJars() throws IOException {
+ java.nio.file.Path tmpDir = temporaryFolder.newFolder().toPath();
+ JobGraph jobGraph = new JobGraph();
+
+ Collection<Path> jars = Arrays.asList(
+ new Path(Files.createFile(tmpDir.resolve("jar1.jar")).toString()),
+ new Path(Files.createFile(tmpDir.resolve("jar2.jar")).toString()));
+
+ jars.forEach(jobGraph::addJar);
+
+ assertEquals(jars.size(), jobGraph.getUserJars().size());
+ assertEquals(0, jobGraph.getUserJarBlobKeys().size());
+
+ try (BlobClient blobClient = new BlobClient(new InetSocketAddress("localhost", blobServer.getPort()), new Configuration())) {
+ ClientUtils.uploadAndSetUserJars(jobGraph, blobClient);
+ }
+
+ assertEquals(jars.size(), jobGraph.getUserJars().size());
+ assertEquals(jars.size(), jobGraph.getUserJarBlobKeys().size());
+ assertEquals(jars.size(), jobGraph.getUserJarBlobKeys().stream().distinct().count());
--- End diff --
I will use `blobServer.getFile()` instead. to verify the validity of the blob keys
---
[GitHub] flink pull request #6199: [FLINK-9624][runtime] Move jar/artifact upload out...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/6199#discussion_r197375181
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/client/ClientUtils.java ---
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.client;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.cache.DistributedCache;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.blob.BlobClient;
+import org.apache.flink.runtime.blob.PermanentBlobKey;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Map;
+
+/**
+ * Contains utility methods for clients.
+ */
+public enum ClientUtils {
+ ;
+
+ /**
+ * Uploads the user jars from the given {@link JobGraph} using the given {@link BlobClient},
+ * and sets the appropriate blobkeys.
+ *
+ * @param jobGraph jobgraph requiring user jars
+ * @param blobClient client to upload jars with
+ * @throws IOException if the upload fails
+ */
+ public static void uploadAndSetUserJars(JobGraph jobGraph, BlobClient blobClient) throws IOException {
+ Collection<PermanentBlobKey> blobKeys = uploadUserJars(jobGraph.getJobID(), jobGraph.getUserJars(), blobClient);
+ setUserJarBlobKeys(blobKeys, jobGraph);
+ }
+
+ private static Collection<PermanentBlobKey> uploadUserJars(JobID jobId, Collection<Path> userJars, BlobClient blobClient) throws IOException {
+ Collection<PermanentBlobKey> blobKeys = new ArrayList<>(userJars.size());
+ for (Path jar : userJars) {
+ final PermanentBlobKey blobKey = blobClient.uploadFile(jobId, jar);
+ blobKeys.add(blobKey);
+ }
+ return blobKeys;
+ }
+
+ private static void setUserJarBlobKeys(Collection<PermanentBlobKey> blobKeys, JobGraph jobGraph) {
+ blobKeys.forEach(jobGraph::addUserJarBlobKey);
+ }
+
+ /**
+ * Uploads the user artifacts from the given {@link JobGraph} using the given {@link BlobClient},
+ * and sets the appropriate blobkeys.
+ *
+ * @param jobGraph jobgraph requiring user artifacts
+ * @param blobClient client to upload artifacts with
+ * @throws IOException if the upload fails
+ */
+ public static void uploadAndSetUserArtifacts(JobGraph jobGraph, BlobClient blobClient) throws IOException {
+ Collection<Tuple2<String, PermanentBlobKey>> blobKeys = uploadUserArtifacts(jobGraph.getJobID(), jobGraph.getUserArtifacts(), blobClient);
+ setUserArtifactBlobKeys(jobGraph, blobKeys);
+ }
+
+ private static Collection<Tuple2<String, PermanentBlobKey>> uploadUserArtifacts(JobID jobID, Map<String, DistributedCache.DistributedCacheEntry> userArtifacts, BlobClient blobClient) throws IOException {
--- End diff --
Signature could be changed to accept a Map<String, Path> instead.
---