flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobClient.java
/**
	 * Reconstructs the class loader by first requesting information about it at the JobManager
	 * and then downloading missing jar files.
	 * @param jobID id of job
	 * @param jobManager gateway to the JobManager
	 * @param config the flink configuration
	 * @return A classloader that should behave like the original classloader
	 * @throws JobRetrievalException if anything goes wrong
	 */
Reconstructs the class loader by first requesting information about it at the JobManager
and then downloading missing jar files.
Reconstructs the class loader by first requesting information about it at the JobManager and then downloading missing jar files .
public static ClassLoader retrieveClassLoader(
			JobID jobID,
			JobManagerGateway jobManager,
			Configuration config,
			HighAvailabilityServices highAvailabilityServices,
			Time timeout)
		throws JobRetrievalException {
		final CompletableFuture<Optional<JobManagerMessages.ClassloadingProps>> clPropsFuture = jobManager
			.requestClassloadingProps(jobID, timeout);
		final Optional<JobManagerMessages.ClassloadingProps> optProps;
		try {
			optProps = clPropsFuture.get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
		} catch (Exception e) {
			throw new JobRetrievalException(jobID, "Could not retrieve the class loading properties from JobManager.", e);
		}
		if (optProps.isPresent()) {
			JobManagerMessages.ClassloadingProps props = optProps.get();
			InetSocketAddress serverAddress = new InetSocketAddress(jobManager.getHostname(), props.blobManagerPort());
			final PermanentBlobCache permanentBlobCache;
			try {
				permanentBlobCache = new PermanentBlobCache(serverAddress, config, highAvailabilityServices.createBlobStore());
			} catch (IOException e) {
				throw new JobRetrievalException(
					jobID,
					"Failed to setup BlobCache.",
					e);
			}
			final Collection<PermanentBlobKey> requiredJarFiles = props.requiredJarFiles();
			final Collection<URL> requiredClasspaths = props.requiredClasspaths();
			final URL[] allURLs = new URL[requiredJarFiles.size() + requiredClasspaths.size()];
			int pos = 0;
			for (PermanentBlobKey blobKey : props.requiredJarFiles()) {
				try {
					allURLs[pos++] = permanentBlobCache.getFile(jobID, blobKey).toURI().toURL();
				} catch (Exception e) {
					try {
						permanentBlobCache.close();
					} catch (IOException ioe) {
						LOG.warn("Could not properly close the BlobClient.", ioe);
					}
					throw new JobRetrievalException(jobID, "Failed to download BlobKey " + blobKey, e);
				}
			}
			for (URL url : requiredClasspaths) {
				allURLs[pos++] = url;
			}
			return FlinkUserCodeClassLoaders.parentFirst(allURLs, JobClient.class.getClassLoader());
		} else {
			throw new JobRetrievalException(jobID, "Couldn't retrieve class loader. Job " + jobID + " not found");
		}
	}
public static ClassLoader retrieveClassLoader ( JobID jobID , JobManagerGateway jobManager , Configuration config , HighAvailabilityServices highAvailabilityServices , Time timeout ) throws JobRetrievalException { final CompletableFuture < Optional < JobManagerMessages . ClassloadingProps > > clPropsFuture = jobManager . requestClassloadingProps ( jobID , timeout ) ; final Optional < JobManagerMessages . ClassloadingProps > optProps ; try { optProps = clPropsFuture . get ( timeout . toMilliseconds ( ) , TimeUnit . MILLISECONDS ) ; } catch ( Exception e ) { throw new JobRetrievalException ( jobID , "Could not retrieve the class loading properties from JobManager." , e ) ; } if ( optProps . isPresent ( ) ) { JobManagerMessages . ClassloadingProps props = optProps . get ( ) ; InetSocketAddress serverAddress = new InetSocketAddress ( jobManager . getHostname ( ) , props . blobManagerPort ( ) ) ; final PermanentBlobCache permanentBlobCache ; try { permanentBlobCache = new PermanentBlobCache ( serverAddress , config , highAvailabilityServices . createBlobStore ( ) ) ; } catch ( IOException e ) { throw new JobRetrievalException ( jobID , "Failed to setup BlobCache." , e ) ; } final Collection < PermanentBlobKey > requiredJarFiles = props . requiredJarFiles ( ) ; final Collection < URL > requiredClasspaths = props . requiredClasspaths ( ) ; final URL [ ] allURLs = new URL [ requiredJarFiles . size ( ) + requiredClasspaths . size ( ) ] ; int pos = 0 ; for ( PermanentBlobKey blobKey : props . requiredJarFiles ( ) ) { try { allURLs [ pos ++ ] = permanentBlobCache . getFile ( jobID , blobKey ) . toURI ( ) . toURL ( ) ; } catch ( Exception e ) { try { permanentBlobCache . close ( ) ; } catch ( IOException ioe ) { LOG . warn ( "Could not properly close the BlobClient." , ioe ) ; } throw new JobRetrievalException ( jobID , "Failed to download BlobKey " + blobKey , e ) ; } } for ( URL url : requiredClasspaths ) { allURLs [ pos ++ ] = url ; } return FlinkUserCodeClassLoaders . parentFirst ( allURLs , JobClient . class . getClassLoader ( ) ) ; } else { throw new JobRetrievalException ( jobID , "Couldn't retrieve class loader. Job " + jobID + " not found" ) ; } }
Reconstructs the class loader by first requesting information about it at the JobManager
 and then downloading missing jar files.
Reconstructs the class loader by first requesting information about it at the JobManager and then downloading missing jar files .
// TODO: Fix lifecycle of PermanentBlobCache to properly close it upon usage