@Test public void testSetUserArtifactBlobKey() throws IOException, ClassNotFoundException { JobGraph jb = new JobGraph(); final DistributedCache.DistributedCacheEntry[] entries = { new DistributedCache.DistributedCacheEntry("p1", true, true), new DistributedCache.DistributedCacheEntry("p2", true, false), new DistributedCache.DistributedCacheEntry("p3", false, true), new DistributedCache.DistributedCacheEntry("p4", true, false), }; for (DistributedCache.DistributedCacheEntry entry : entries) { jb.addUserArtifact(entry.filePath, entry); } for (DistributedCache.DistributedCacheEntry entry : entries) { PermanentBlobKey blobKey = new PermanentBlobKey(); jb.setUserArtifactBlobKey(entry.filePath, blobKey); DistributedCache.DistributedCacheEntry jobGraphEntry = jb.getUserArtifacts().get(entry.filePath); assertNotNull(jobGraphEntry); assertEquals(blobKey, InstantiationUtil.deserializeObject(jobGraphEntry.blobKey, ClassLoader.getSystemClassLoader(), false)); assertEquals(entry.isExecutable, jobGraphEntry.isExecutable); assertEquals(entry.isZipped, jobGraphEntry.isZipped); assertEquals(entry.filePath, jobGraphEntry.filePath); } }
public void setUserArtifactBlobKey(String entryName, PermanentBlobKey blobKey) throws IOException { byte[] serializedBlobKey; serializedBlobKey = InstantiationUtil.serializeObject(blobKey); userArtifacts.computeIfPresent(entryName, (key, originalEntry) -> new DistributedCache.DistributedCacheEntry( originalEntry.filePath, originalEntry.isExecutable, serializedBlobKey, originalEntry.isZipped )); }
JobGraph implements Serializable { public void setUserArtifactBlobKey(String entryName, PermanentBlobKey blobKey) throws IOException { byte[] serializedBlobKey; serializedBlobKey = InstantiationUtil.serializeObject(blobKey); userArtifacts.computeIfPresent(entryName, (key, originalEntry) -> new DistributedCache.DistributedCacheEntry( originalEntry.filePath, originalEntry.isExecutable, serializedBlobKey, originalEntry.isZipped )); } }
JobGraph implements Serializable { public void setUserArtifactBlobKey(String entryName, PermanentBlobKey blobKey) throws IOException { byte[] serializedBlobKey; serializedBlobKey = InstantiationUtil.serializeObject(blobKey); userArtifacts.computeIfPresent(entryName, (key, originalEntry) -> new DistributedCache.DistributedCacheEntry( originalEntry.filePath, originalEntry.isExecutable, serializedBlobKey, originalEntry.isZipped )); } JobGraph(String jobName);  JobGraph(JobID jobId, String jobName);  JobGraph(JobVertex... vertices);  JobGraph(String jobName, JobVertex... vertices);  JobGraph(JobID jobId, String jobName, JobVertex... vertices); }
JobGraph implements Serializable { public void setUserArtifactBlobKey(String entryName, PermanentBlobKey blobKey) throws IOException { byte[] serializedBlobKey; serializedBlobKey = InstantiationUtil.serializeObject(blobKey); userArtifacts.computeIfPresent(entryName, (key, originalEntry) -> new DistributedCache.DistributedCacheEntry( originalEntry.filePath, originalEntry.isExecutable, serializedBlobKey, originalEntry.isZipped )); } JobGraph(String jobName);  JobGraph(JobID jobId, String jobName);  JobGraph(JobVertex... vertices);  JobGraph(String jobName, JobVertex... vertices);  JobGraph(JobID jobId, String jobName, JobVertex... vertices); JobID getJobID(); String getName(); Configuration getJobConfiguration(); SerializedValue<ExecutionConfig> getSerializedExecutionConfig(); long getSessionTimeout(); void setSessionTimeout(long sessionTimeout); void setAllowQueuedScheduling(boolean allowQueuedScheduling); boolean getAllowQueuedScheduling(); void setScheduleMode(ScheduleMode scheduleMode); ScheduleMode getScheduleMode(); void setSavepointRestoreSettings(SavepointRestoreSettings settings); SavepointRestoreSettings getSavepointRestoreSettings(); void setExecutionConfig(ExecutionConfig executionConfig); void addVertex(JobVertex vertex); Iterable<JobVertex> getVertices(); JobVertex[] getVerticesAsArray(); int getNumberOfVertices(); void setSnapshotSettings(JobCheckpointingSettings settings); JobCheckpointingSettings getCheckpointingSettings(); boolean isCheckpointingEnabled(); JobVertex findVertexByID(JobVertexID id); void setClasspaths(List<URL> paths); List<URL> getClasspaths(); int getMaximumParallelism(); List<JobVertex> getVerticesSortedTopologicallyFromSources(); void addJar(Path jar); List<Path> getUserJars(); void addUserArtifact(String name, DistributedCache.DistributedCacheEntry file); Map<String, DistributedCache.DistributedCacheEntry> getUserArtifacts(); void addUserJarBlobKey(PermanentBlobKey key); boolean hasUsercodeJarFiles(); List<PermanentBlobKey> getUserJarBlobKeys(); @Override String toString(); void setUserArtifactBlobKey(String entryName, PermanentBlobKey blobKey); void writeUserArtifactEntriesToConfiguration(); }
JobGraph implements Serializable { public void setUserArtifactBlobKey(String entryName, PermanentBlobKey blobKey) throws IOException { byte[] serializedBlobKey; serializedBlobKey = InstantiationUtil.serializeObject(blobKey); userArtifacts.computeIfPresent(entryName, (key, originalEntry) -> new DistributedCache.DistributedCacheEntry( originalEntry.filePath, originalEntry.isExecutable, serializedBlobKey, originalEntry.isZipped )); } JobGraph(String jobName);  JobGraph(JobID jobId, String jobName);  JobGraph(JobVertex... vertices);  JobGraph(String jobName, JobVertex... vertices);  JobGraph(JobID jobId, String jobName, JobVertex... vertices); JobID getJobID(); String getName(); Configuration getJobConfiguration(); SerializedValue<ExecutionConfig> getSerializedExecutionConfig(); long getSessionTimeout(); void setSessionTimeout(long sessionTimeout); void setAllowQueuedScheduling(boolean allowQueuedScheduling); boolean getAllowQueuedScheduling(); void setScheduleMode(ScheduleMode scheduleMode); ScheduleMode getScheduleMode(); void setSavepointRestoreSettings(SavepointRestoreSettings settings); SavepointRestoreSettings getSavepointRestoreSettings(); void setExecutionConfig(ExecutionConfig executionConfig); void addVertex(JobVertex vertex); Iterable<JobVertex> getVertices(); JobVertex[] getVerticesAsArray(); int getNumberOfVertices(); void setSnapshotSettings(JobCheckpointingSettings settings); JobCheckpointingSettings getCheckpointingSettings(); boolean isCheckpointingEnabled(); JobVertex findVertexByID(JobVertexID id); void setClasspaths(List<URL> paths); List<URL> getClasspaths(); int getMaximumParallelism(); List<JobVertex> getVerticesSortedTopologicallyFromSources(); void addJar(Path jar); List<Path> getUserJars(); void addUserArtifact(String name, DistributedCache.DistributedCacheEntry file); Map<String, DistributedCache.DistributedCacheEntry> getUserArtifacts(); void addUserJarBlobKey(PermanentBlobKey key); boolean hasUsercodeJarFiles(); List<PermanentBlobKey> getUserJarBlobKeys(); @Override String toString(); void setUserArtifactBlobKey(String entryName, PermanentBlobKey blobKey); void writeUserArtifactEntriesToConfiguration();  }