[FLINK-1968] [runtime] Clean up and improve the distributed cache.

 - Gives a proper exception when a non-cached file is accessed
 - Forwards I/O exceptions that happen during file transfer, rather than only returning null when transfer failed
 - Consistently keeps reference counts and copies only when needed
 - Properly removes all files when shutdown
 - Uses a shutdown hook to remove files when process is killed

flink-core/src/main/java/org/apache/flink/api/common/cache/DistributedCache.java

@@ -21,11 +21,13 @@
 
 
 import java.io.File;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Set;
-import java.util.concurrent.FutureTask;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
 
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.core.fs.Path;
@@ -37,6 +39,7 @@
 public class DistributedCache {
 
 	public static class DistributedCacheEntry {
+
 		public String filePath;
 		public Boolean isExecutable;
 
@@ -45,19 +48,45 @@ public DistributedCacheEntry(String filePath, Boolean isExecutable){
 			this.isExecutable=isExecutable;
 		}
 	}
-
-	final static String CACHE_FILE_NUM = "DISTRIBUTED_CACHE_FILE_NUM";
-
-	final static String CACHE_FILE_NAME = "DISTRIBUTED_CACHE_FILE_NAME_";
-
-	final static String CACHE_FILE_PATH = "DISTRIBUTED_CACHE_FILE_PATH_";
 
-	final static String CACHE_FILE_EXE = "DISTRIBUTED_CACHE_FILE_EXE_";
+	// ------------------------------------------------------------------------
 
-	public final static String TMP_PREFIX = "tmp_";
+	private final Map<String, Future<Path>> cacheCopyTasks;
 
-	private Map<String, FutureTask<Path>> cacheCopyTasks = new HashMap<String, FutureTask<Path>>();
+	public DistributedCache(Map<String, Future<Path>> cacheCopyTasks) {
+		this.cacheCopyTasks = cacheCopyTasks;
+	}
+
+	// ------------------------------------------------------------------------
 
+	public File getFile(String name) {
+		if (name == null) {
+			throw new NullPointerException("name must not be null");
+		}
+
+		Future<Path> future = cacheCopyTasks.get(name);
+		if (future == null) {
+			throw new IllegalArgumentException("File with name '" + name + "' is not available." +
+					" Did you forget to register the file?");
+		}
+
+		try {
+			Path tmp = future.get();
+			return new File(tmp.toString());
+		}
+		catch (ExecutionException e) {
+			throw new RuntimeException("An error occurred while copying the file.", e.getCause());
+		}
+		catch (Exception e) {
+			throw new RuntimeException("Error while getting the file registered under '" + name +
+					"' from the distributed cache", e);
+		}
+	}
+
+	// ------------------------------------------------------------------------
+	//  Utilities to read/write cache files from/to the configuration
+	// ------------------------------------------------------------------------
+
 	public static void writeFileInfoToConfig(String name, DistributedCacheEntry e, Configuration conf) {
 		int num = conf.getInteger(CACHE_FILE_NUM,0) + 1;
 		conf.setInteger(CACHE_FILE_NUM, num);
@@ -67,29 +96,26 @@ public static void writeFileInfoToConfig(String name, DistributedCacheEntry e, C
 	}
 
 	public static Set<Entry<String, DistributedCacheEntry>> readFileInfoFromConfig(Configuration conf) {
-		Map<String, DistributedCacheEntry> cacheFiles = new HashMap<String, DistributedCacheEntry>();
 		int num = conf.getInteger(CACHE_FILE_NUM, 0);
+		if (num == 0) {
+			return Collections.emptySet();
+		}
+
+		Map<String, DistributedCacheEntry> cacheFiles = new HashMap<String, DistributedCacheEntry>();
 		for (int i = 1; i <= num; i++) {
-			String name = conf.getString(CACHE_FILE_NAME + i, "");
-			String filePath = conf.getString(CACHE_FILE_PATH + i, "");
+			String name = conf.getString(CACHE_FILE_NAME + i, null);
+			String filePath = conf.getString(CACHE_FILE_PATH + i, null);
 			Boolean isExecutable = conf.getBoolean(CACHE_FILE_EXE + i, false);
 			cacheFiles.put(name, new DistributedCacheEntry(filePath, isExecutable));
 		}
 		return cacheFiles.entrySet();
 	}
 
-	public void setCopyTasks(Map<String, FutureTask<Path>> cpTasks) {
-			this.cacheCopyTasks = cpTasks;
-	}
+	private static final String CACHE_FILE_NUM = "DISTRIBUTED_CACHE_FILE_NUM";
 
-	public File getFile(String name) {
-		Path tmp = null;
-		//The FutureTask.get() method will block until the file is ready.
-		try {
-			tmp = cacheCopyTasks.get(name).get();
-		} catch (Exception  e) {
-			throw new RuntimeException("Error while getting file from distributed cache", e);
-		}
-		return new File(tmp.toString());
-	}
+	private static final String CACHE_FILE_NAME = "DISTRIBUTED_CACHE_FILE_NAME_";
+
+	private static final String CACHE_FILE_PATH = "DISTRIBUTED_CACHE_FILE_PATH_";
+
+	private static final String CACHE_FILE_EXE = "DISTRIBUTED_CACHE_FILE_EXE_";
 }

flink-core/src/main/java/org/apache/flink/api/common/functions/util/AbstractRuntimeUDFContext.java

@@ -19,9 +19,10 @@
 package org.apache.flink.api.common.functions.util;
 
 import java.io.Serializable;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
-import java.util.concurrent.FutureTask;
+import java.util.concurrent.Future;
 
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.accumulators.Accumulator;
@@ -51,20 +52,30 @@ public abstract class AbstractRuntimeUDFContext implements RuntimeContext {
 
 	private final HashMap<String, Accumulator<?, ?>> accumulators = new HashMap<String, Accumulator<?, ?>>();
 
-	private final DistributedCache distributedCache = new DistributedCache();
+	private final DistributedCache distributedCache;
 
 
-	public AbstractRuntimeUDFContext(String name, int numParallelSubtasks, int subtaskIndex, ClassLoader userCodeClassLoader, ExecutionConfig executionConfig) {
+	public AbstractRuntimeUDFContext(String name,
+										int numParallelSubtasks, int subtaskIndex,
+										ClassLoader userCodeClassLoader,
+										ExecutionConfig executionConfig)
+	{
+		this(name, numParallelSubtasks, subtaskIndex, userCodeClassLoader, executionConfig,
+				Collections.<String, Future<Path>>emptyMap());
+	}
+
+	public AbstractRuntimeUDFContext(String name,
+										int numParallelSubtasks, int subtaskIndex,
+										ClassLoader userCodeClassLoader,
+										ExecutionConfig executionConfig,
+										Map<String, Future<Path>> cpTasks)
+	{
 		this.name = name;
 		this.numParallelSubtasks = numParallelSubtasks;
 		this.subtaskIndex = subtaskIndex;
 		this.userCodeClassLoader = userCodeClassLoader;
 		this.executionConfig = executionConfig;
-	}
-
-	public AbstractRuntimeUDFContext(String name, int numParallelSubtasks, int subtaskIndex, ClassLoader userCodeClassLoader, ExecutionConfig executionConfig, Map<String, FutureTask<Path>> cpTasks) {
-		this(name, numParallelSubtasks, subtaskIndex, userCodeClassLoader, executionConfig);
-		this.distributedCache.setCopyTasks(cpTasks);
+		this.distributedCache = new DistributedCache(cpTasks);
 	}
 
 	@Override

flink-core/src/main/java/org/apache/flink/api/common/functions/util/RuntimeUDFContext.java

@@ -21,7 +21,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.FutureTask;
+import java.util.concurrent.Future;
 
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.functions.BroadcastVariableInitializer;
@@ -42,7 +42,7 @@ public RuntimeUDFContext(String name, int numParallelSubtasks, int subtaskIndex,
 		super(name, numParallelSubtasks, subtaskIndex, userCodeClassLoader, executionConfig);
 	}
 
-	public RuntimeUDFContext(String name, int numParallelSubtasks, int subtaskIndex, ClassLoader userCodeClassLoader, ExecutionConfig executionConfig, Map<String, FutureTask<Path>> cpTasks) {
+	public RuntimeUDFContext(String name, int numParallelSubtasks, int subtaskIndex, ClassLoader userCodeClassLoader, ExecutionConfig executionConfig, Map<String, Future<Path>> cpTasks) {
 		super(name, numParallelSubtasks, subtaskIndex, userCodeClassLoader, executionConfig, cpTasks);
 	}
 

flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java

@@ -168,17 +168,15 @@ public int hashCode() {
 	 * @throws IOException
 	 *         thrown if a reference to the file system instance could not be obtained
 	 */
-	public static FileSystem getLocalFileSystem() throws IOException {
-
-		URI localUri;
-
+	public static FileSystem getLocalFileSystem() {
+		// this should really never fail.
 		try {
-			localUri = OperatingSystem.isWindows() ?  new URI("file:/") : new URI("file:///");
-		} catch (URISyntaxException e) {
-			throw new IOException("Cannot create URI for local file system");
+			URI localUri = OperatingSystem.isWindows() ? new URI("file:/") : new URI("file:///");
+			return get(localUri);
+		}
+		catch (Exception e) {
+			throw new RuntimeException("Cannot create URI for local file system");
 		}
-
-		return get(localUri);
 	}
 
 	/**
@@ -193,7 +191,7 @@ public static FileSystem getLocalFileSystem() throws IOException {
 	 *         thrown if a reference to the file system instance could not be obtained
 	 */
 	public static FileSystem get(URI uri) throws IOException {
-		FileSystem fs = null;
+		FileSystem fs;
 
 		synchronized (SYNCHRONIZATION_OBJECT) {
 

flink-runtime/src/main/java/org/apache/flink/runtime/execution/Environment.java

@@ -18,58 +18,62 @@
 
 package org.apache.flink.runtime.execution;
 
-import akka.actor.ActorRef;
 import org.apache.flink.api.common.accumulators.Accumulator;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.core.fs.Path;
 import org.apache.flink.runtime.broadcast.BroadcastVariableManager;
+import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
 import org.apache.flink.runtime.io.disk.iomanager.IOManager;
 import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter;
 import org.apache.flink.runtime.io.network.partition.consumer.InputGate;
-import org.apache.flink.runtime.jobgraph.JobGraph;
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.runtime.jobgraph.JobVertexID;
 import org.apache.flink.runtime.jobgraph.tasks.InputSplitProvider;
 import org.apache.flink.runtime.memorymanager.MemoryManager;
 
 import java.util.Map;
-import java.util.concurrent.FutureTask;
+import java.util.concurrent.Future;
 
 /**
- * The user code of every task runs inside an <code>Environment</code> object.
- * The environment provides important services to the task. It keeps track of
- * setting up the communication channels and provides access to input splits,
- * memory manager, etc.
+ * The Environment gives the code executed in a task access to the task's properties
+ * (such as name, parallelism), the configurations, the data stream readers and writers,
+ * as well as the various components that are provided by the TaskManager, such as
+ * memory manager, I/O manager, ...
  */
 public interface Environment {
 
 	/**
-	 * Returns the ID of the job from the original job graph. It is used by the library cache manager to find the
-	 * required
-	 * libraries for executing the assigned Nephele task.
+	 * Returns the ID of the job that the task belongs to.
 	 *
 	 * @return the ID of the job from the original job graph
 	 */
 	JobID getJobID();
 
 	/**
-	 * Gets the ID of the jobVertex that this task corresponds to.
+	 * Gets the ID of the JobVertex for which this task executes a parallel subtask.
 	 *
 	 * @return The JobVertexID of this task.
 	 */
 	JobVertexID getJobVertexId();
 
 	/**
-	 * Returns the task configuration object which was attached to the original JobVertex.
+	 * Gets the ID of the task execution attempt.
 	 *
-	 * @return the task configuration object which was attached to the original JobVertex.
+	 * @return The ID of the task execution attempt.
+	 */
+	ExecutionAttemptID getExecutionId();
+
+	/**
+	 * Returns the task-wide configuration object, originally attache to the job vertex.
+	 *
+	 * @return The task-wide configuration
 	 */
 	Configuration getTaskConfiguration();
 
 	/**
-	 * Returns the job configuration object which was attached to the original {@link JobGraph}.
+	 * Returns the job-wide configuration object that was attached to the JobGraph.
 	 *
-	 * @return the job configuration object which was attached to the original {@link JobGraph}
+	 * @return The job-wide configuration
 	 */
 	Configuration getJobConfiguration();
 
@@ -132,7 +136,7 @@ public interface Environment {
 	 */
 	ClassLoader getUserClassLoader();
 
-	Map<String, FutureTask<Path>> getCopyTask();
+	Map<String, Future<Path>> getDistributedCacheEntries();
 
 	BroadcastVariableManager getBroadcastVariableManager();
 
@@ -155,13 +159,4 @@ public interface Environment {
 
 	InputGate[] getAllInputGates();
 
-
-	/**
-	 * Returns the proxy object for the accumulator protocol.
-	 */
-	// THIS DOES NOT BELONG HERE, THIS TOTALLY BREAKS COMPONENTIZATION.
-	// THE EXECUTED TASKS HAVE BEEN KEPT INDEPENDENT OF ANY RPC OR ACTOR
-	// COMMUNICATION !!!
-	ActorRef getJobManager();
-
 }