[06/14] flink git commit: [FLINK-7068][blob] change BlobService sub-classes for permanent and transient BLOBs

[tr...@apache.org]

http://git-wip-us.apache.org/repos/asf/flink/blob/071e27f7/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheDeleteTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheDeleteTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheDeleteTest.java
new file mode 100644
index 0000000..20b0152
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheDeleteTest.java
@@ -0,0 +1,413 @@
+/*
+ * 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.blob;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.configuration.BlobServerOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.concurrent.FutureUtils;
+import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.OperatingSystem;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import javax.annotation.Nullable;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import static org.apache.flink.runtime.blob.BlobServerDeleteTest.delete;
+import static org.apache.flink.runtime.blob.BlobServerGetTest.verifyDeleted;
+import static org.apache.flink.runtime.blob.BlobServerPutTest.put;
+import static org.apache.flink.runtime.blob.BlobServerPutTest.verifyContents;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assume.assumeTrue;
+
+/**
+ * Tests how DELETE requests behave.
+ */
+public class BlobCacheDeleteTest extends TestLogger {
+
+	private final Random rnd = new Random();
+
+	@Rule
+	public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+	@Test
+	public void testDeleteTransient1() throws IOException {
+		testDeleteTransient(null, new JobID());
+	}
+
+	@Test
+	public void testDeleteTransient2() throws IOException {
+		testDeleteTransient(new JobID(), null);
+	}
+
+	@Test
+	public void testDeleteTransient3() throws IOException {
+		testDeleteTransient(null, null);
+	}
+
+	@Test
+	public void testDeleteTransient4() throws IOException {
+		testDeleteTransient(new JobID(), new JobID());
+	}
+
+	/**
+	 * Uploads a (different) byte array for each of the given jobs and verifies that deleting one of
+	 * them (via the {@link BlobCache}) does not influence the other.
+	 *
+	 * @param jobId1
+	 * 		first job id
+	 * @param jobId2
+	 * 		second job id
+	 */
+	private void testDeleteTransient(@Nullable JobID jobId1, @Nullable JobID jobId2)
+			throws IOException {
+
+		final Configuration config = new Configuration();
+		config.setString(BlobServerOptions.STORAGE_DIRECTORY, temporaryFolder.newFolder().getAbsolutePath());
+
+		try (
+			BlobServer server = new BlobServer(config, new VoidBlobStore());
+			BlobCache cache = new BlobCache(new InetSocketAddress("localhost", server.getPort()),
+				config, new VoidBlobStore())) {
+
+			server.start();
+
+			byte[] data = new byte[2000000];
+			rnd.nextBytes(data);
+			byte[] data2 = Arrays.copyOf(data, data.length);
+			data2[0] ^= 1;
+
+			// put first BLOB
+			BlobKey key1 = put(server, jobId1, data, false);
+			assertNotNull(key1);
+
+			// put two more BLOBs (same key, other key) for another job ID
+			BlobKey key2a = put(server, jobId2, data, false);
+			assertNotNull(key2a);
+			assertEquals(key1, key2a);
+			BlobKey key2b = put(server, jobId2, data2, false);
+			assertNotNull(key2b);
+
+			// issue a DELETE request
+			assertTrue(delete(cache, jobId1, key1));
+
+			verifyDeleted(cache, jobId1, key1, false);
+			verifyDeleted(server, jobId1, key1, false);
+			// deleting a one BLOB should not affect another BLOB, even with the same key if job IDs are different
+			if ((jobId1 == null && jobId2 != null) || (jobId1 != null && !jobId1.equals(jobId2))) {
+				verifyContents(server, jobId2, key2a, data, false);
+			}
+			verifyContents(server, jobId2, key2b, data2, false);
+
+			// delete first file of second job
+			assertTrue(delete(cache, jobId2, key2a));
+			verifyDeleted(cache, jobId2, key2a, false);
+			verifyDeleted(server, jobId2, key2a, false);
+			verifyContents(server, jobId2, key2b, data2, false);
+
+			// delete second file of second job
+			assertTrue(delete(cache, jobId2, key2a));
+			verifyDeleted(cache, jobId2, key2a, false);
+			verifyDeleted(server, jobId2, key2a, false);
+			verifyContents(server, jobId2, key2b, data2, false);
+		}
+	}
+
+	@Test
+	public void testDeleteTransientAlreadyDeletedNoJob() throws IOException {
+		testDeleteTransientAlreadyDeleted(null);
+	}
+
+	@Test
+	public void testDeleteTransientAlreadyDeletedForJob() throws IOException {
+		testDeleteTransientAlreadyDeleted(new JobID());
+	}
+
+	/**
+	 * Uploads a byte array for the given job and verifies that deleting it (via the {@link
+	 * BlobCache}) does not fail independent of whether the file exists.
+	 *
+	 * @param jobId
+	 * 		job id
+	 */
+	private void testDeleteTransientAlreadyDeleted(@Nullable final JobID jobId) throws IOException {
+
+		final Configuration config = new Configuration();
+		config.setString(BlobServerOptions.STORAGE_DIRECTORY, temporaryFolder.newFolder().getAbsolutePath());
+
+		try (
+			BlobServer server = new BlobServer(config, new VoidBlobStore());
+			BlobCache cache = new BlobCache(new InetSocketAddress("localhost", server.getPort()),
+				config, new VoidBlobStore())) {
+
+			server.start();
+
+			byte[] data = new byte[2000000];
+			rnd.nextBytes(data);
+
+			// put BLOB
+			BlobKey key = put(server, jobId, data, false);
+			assertNotNull(key);
+
+			File blobFile = server.getStorageLocation(jobId, key);
+			assertTrue(blobFile.delete());
+
+			// DELETE operation should not fail if file is already deleted
+			assertTrue(delete(cache, jobId, key));
+			verifyDeleted(cache, jobId, key, false);
+
+			// one more delete call that should not fail
+			assertTrue(delete(cache, jobId, key));
+			verifyDeleted(cache, jobId, key, false);
+		}
+	}
+
+	@Test
+	public void testDeleteTransientLocalFailsNoJob() throws IOException {
+		testDeleteTransientLocalFails(null);
+	}
+
+	@Test
+	public void testDeleteTransientLocalFailsForJob() throws IOException {
+		testDeleteTransientLocalFails(new JobID());
+	}
+
+	/**
+	 * Uploads a byte array for the given job and verifies that a delete operation (via the {@link
+	 * BlobCache}) does not fail even if the file is not deletable locally, e.g. via restricting
+	 * the permissions.
+	 *
+	 * @param jobId
+	 * 		job id
+	 */
+	private void testDeleteTransientLocalFails(@Nullable final JobID jobId) throws IOException {
+		assumeTrue(!OperatingSystem.isWindows()); //setWritable doesn't work on Windows.
+
+		final Configuration config = new Configuration();
+		config.setString(BlobServerOptions.STORAGE_DIRECTORY, temporaryFolder.newFolder().getAbsolutePath());
+
+		File blobFile = null;
+		File directory = null;
+		try (
+			BlobServer server = new BlobServer(config, new VoidBlobStore());
+			BlobCache cache = new BlobCache(new InetSocketAddress("localhost", server.getPort()),
+				config, new VoidBlobStore())) {
+
+			server.start();
+
+			try {
+				byte[] data = new byte[2000000];
+				rnd.nextBytes(data);
+
+				// put BLOB
+				BlobKey key = put(server, jobId, data, false);
+				assertNotNull(key);
+
+				// access from cache once to have it available there
+				verifyContents(cache, jobId, key, data, false);
+
+				blobFile = cache.getTransientBlobStore().getStorageLocation(jobId, key);
+				directory = blobFile.getParentFile();
+
+				assertTrue(blobFile.setWritable(false, false));
+				assertTrue(directory.setWritable(false, false));
+
+				// issue a DELETE request
+				assertFalse(delete(cache, jobId, key));
+				verifyDeleted(server, jobId, key, false);
+
+				// the file should still be there on both cache and server
+				verifyContents(cache, jobId, key, data, false);
+			} finally {
+				if (blobFile != null && directory != null) {
+					//noinspection ResultOfMethodCallIgnored
+					blobFile.setWritable(true, false);
+					//noinspection ResultOfMethodCallIgnored
+					directory.setWritable(true, false);
+				}
+			}
+		}
+	}
+
+	@Test
+	public void testDeleteTransientRemoteFailsNoJob() throws IOException {
+		testDeleteTransientRemoteFails(null);
+	}
+
+	@Test
+	public void testDeleteTransientRemoteFailsForJob() throws IOException {
+		testDeleteTransientRemoteFails(new JobID());
+	}
+
+	/**
+	 * Uploads a byte array for the given job and verifies that a delete operation (via the {@link
+	 * BlobCache}) does not fail even if the file is not deletable on the {@link BlobServer}, e.g.
+	 * via restricting the permissions.
+	 *
+	 * @param jobId
+	 * 		job id
+	 */
+	private void testDeleteTransientRemoteFails(@Nullable final JobID jobId) throws IOException {
+		assumeTrue(!OperatingSystem.isWindows()); //setWritable doesn't work on Windows.
+
+		final Configuration config = new Configuration();
+		config.setString(BlobServerOptions.STORAGE_DIRECTORY, temporaryFolder.newFolder().getAbsolutePath());
+
+		File blobFile = null;
+		File directory = null;
+
+		try (
+			BlobServer server = new BlobServer(config, new VoidBlobStore());
+			BlobCache cache = new BlobCache(new InetSocketAddress("localhost", server.getPort()),
+				config, new VoidBlobStore())) {
+
+			server.start();
+
+			try {
+				byte[] data = new byte[2000000];
+				rnd.nextBytes(data);
+
+				// put BLOB
+				BlobKey key = put(server, jobId, data, false);
+				assertNotNull(key);
+
+				// access from cache once to have it available there
+				verifyContents(cache, jobId, key, data, false);
+
+				blobFile = server.getStorageLocation(jobId, key);
+				directory = blobFile.getParentFile();
+
+				assertTrue(blobFile.setWritable(false, false));
+				assertTrue(directory.setWritable(false, false));
+
+				// issue a DELETE request
+				assertFalse(delete(cache, jobId, key));
+				File blobFileAtCache = cache.getTransientBlobStore().getStorageLocation(jobId, key);
+				assertFalse(blobFileAtCache.exists());
+
+				// the file should still be there on the server
+				verifyContents(server, jobId, key, data, false);
+				// ... and may be retrieved by the cache
+				verifyContents(cache, jobId, key, data, false);
+			} finally {
+				if (blobFile != null && directory != null) {
+					//noinspection ResultOfMethodCallIgnored
+					blobFile.setWritable(true, false);
+					//noinspection ResultOfMethodCallIgnored
+					directory.setWritable(true, false);
+				}
+			}
+		}
+	}
+
+	@Test
+	public void testConcurrentDeleteOperationsNoJob() throws IOException, ExecutionException, InterruptedException {
+		testConcurrentDeleteOperations(null);
+	}
+
+	@Test
+	public void testConcurrentDeleteOperationsForJob() throws IOException, ExecutionException, InterruptedException {
+		testConcurrentDeleteOperations(new JobID());
+	}
+
+	/**
+	 * [FLINK-6020] Tests that concurrent delete operations don't interfere with each other.
+	 *
+	 * <p>Note: This test checks that there cannot be two threads which have checked whether a given
+	 * blob file exist and then one of them fails deleting it. Without the introduced lock, this
+	 * situation should rarely happen and make this test fail. Thus, if this test should become
+	 * "unstable", then the delete atomicity is most likely broken.
+	 *
+	 * @param jobId
+	 * 		job ID to use (or <tt>null</tt> if job-unrelated)
+	 */
+	private void testConcurrentDeleteOperations(@Nullable final JobID jobId)
+			throws IOException, InterruptedException, ExecutionException {
+
+		final Configuration config = new Configuration();
+		config.setString(BlobServerOptions.STORAGE_DIRECTORY, temporaryFolder.newFolder().getAbsolutePath());
+
+		final int concurrentDeleteOperations = 3;
+		final ExecutorService executor = Executors.newFixedThreadPool(concurrentDeleteOperations);
+
+		final List<CompletableFuture<Void>> deleteFutures = new ArrayList<>(concurrentDeleteOperations);
+
+		final byte[] data = {1, 2, 3};
+
+		try (
+			BlobServer server = new BlobServer(config, new VoidBlobStore());
+			BlobCache cache = new BlobCache(new InetSocketAddress("localhost", server.getPort()),
+				config, new VoidBlobStore())) {
+
+			server.start();
+
+			final BlobKey blobKey = put(server, jobId, data, false);
+
+			assertTrue(server.getStorageLocation(jobId, blobKey).exists());
+
+			for (int i = 0; i < concurrentDeleteOperations; i++) {
+				CompletableFuture<Void> deleteFuture = CompletableFuture
+					.supplyAsync(
+						() -> {
+							try {
+								assertTrue(delete(cache, jobId, blobKey));
+								assertFalse(cache.getTransientBlobStore().getStorageLocation(jobId, blobKey).exists());
+								assertFalse(server.getStorageLocation(jobId, blobKey).exists());
+								return null;
+							} catch (IOException e) {
+								throw new CompletionException(new FlinkException(
+									"Could not upload blob.", e));
+							}
+						}, executor);
+
+				deleteFutures.add(deleteFuture);
+			}
+
+			CompletableFuture<Void> waitFuture = FutureUtils.waitForAll(deleteFutures);
+
+			// make sure all delete operation have completed successfully
+			// in case of no lock, one of the delete operations should eventually fail
+			waitFuture.get();
+
+			assertFalse(server.getStorageLocation(jobId, blobKey).exists());
+
+		} finally {
+			executor.shutdownNow();
+		}
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/071e27f7/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheGetTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheGetTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheGetTest.java
new file mode 100644
index 0000000..52adc4d
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheGetTest.java
@@ -0,0 +1,490 @@
+/*
+ * 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.blob;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.configuration.BlobServerOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.concurrent.FutureUtils;
+import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.OperatingSystem;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.rules.TemporaryFolder;
+
+import javax.annotation.Nullable;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.file.AccessDeniedException;
+import java.security.MessageDigest;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import static org.apache.flink.runtime.blob.BlobClientTest.validateGetAndClose;
+import static org.apache.flink.runtime.blob.BlobServerGetTest.get;
+import static org.apache.flink.runtime.blob.BlobServerGetTest.verifyDeleted;
+import static org.apache.flink.runtime.blob.BlobServerPutTest.put;
+import static org.apache.flink.runtime.blob.BlobUtils.JOB_DIR_PREFIX;
+import static org.apache.flink.runtime.blob.BlobUtils.NO_JOB_DIR_PREFIX;
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assume.assumeTrue;
+import static org.mockito.Mockito.mock;
+
+/**
+ * Tests how failing GET requests behave in the presence of failures when used with a {@link
+ * BlobCache}.
+ *
+ * <p>Successful GET requests are tested in conjunction wit the PUT requests.
+ */
+public class BlobCacheGetTest extends TestLogger {
+
+	private final Random rnd = new Random();
+
+	@Rule
+	public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+	@Rule
+	public final ExpectedException exception = ExpectedException.none();
+
+	@Test
+	public void testGetFailsDuringLookup1() throws IOException {
+		testGetFailsDuringLookup(null, new JobID(), false);
+	}
+
+	@Test
+	public void testGetFailsDuringLookup2() throws IOException {
+		testGetFailsDuringLookup(new JobID(), new JobID(), false);
+	}
+
+	@Test
+	public void testGetFailsDuringLookup3() throws IOException {
+		testGetFailsDuringLookup(new JobID(), null, false);
+	}
+
+	@Test
+	public void testGetFailsDuringLookupHa() throws IOException {
+		testGetFailsDuringLookup(new JobID(), new JobID(), true);
+	}
+
+	/**
+	 * Checks the correct result if a GET operation fails during the lookup of the file.
+	 *
+	 * @param jobId1 first job ID or <tt>null</tt> if job-unrelated
+	 * @param jobId2 second job ID different to <tt>jobId1</tt>
+	 */
+	private void testGetFailsDuringLookup(final JobID jobId1, final JobID jobId2, boolean highAvailability)
+		throws IOException {
+		final Configuration config = new Configuration();
+		config.setString(BlobServerOptions.STORAGE_DIRECTORY, temporaryFolder.newFolder().getAbsolutePath());
+
+		try (
+			BlobServer server = new BlobServer(config, new VoidBlobStore());
+			BlobCache cache = new BlobCache(new InetSocketAddress("localhost", server.getPort()),
+				config, new VoidBlobStore())) {
+
+			server.start();
+
+			byte[] data = new byte[2000000];
+			rnd.nextBytes(data);
+
+			// put content addressable (like libraries)
+			BlobKey key = put(server, jobId1, data, highAvailability);
+			assertNotNull(key);
+
+			// delete file to make sure that GET requests fail
+			File blobFile = server.getStorageLocation(jobId1, key);
+			assertTrue(blobFile.delete());
+
+			// issue a GET request that fails
+			verifyDeleted(cache, jobId1, key, highAvailability);
+
+			// add the same data under a second jobId
+			BlobKey key2 = put(server, jobId2, data, highAvailability);
+			assertNotNull(key);
+			assertEquals(key, key2);
+
+			// request for jobId2 should succeed
+			get(cache, jobId2, key, highAvailability);
+			// request for jobId1 should still fail
+			verifyDeleted(cache, jobId1, key, highAvailability);
+
+			// delete on cache, try to retrieve again
+			if (highAvailability) {
+				blobFile = cache.getPermanentBlobStore().getStorageLocation(jobId2, key);
+			} else {
+				blobFile = cache.getTransientBlobStore().getStorageLocation(jobId2, key);
+			}
+			assertTrue(blobFile.delete());
+			get(cache, jobId2, key, highAvailability);
+
+			// delete on cache and server, verify that it is not accessible anymore
+			if (highAvailability) {
+				blobFile = cache.getPermanentBlobStore().getStorageLocation(jobId2, key);
+			} else {
+				blobFile = cache.getTransientBlobStore().getStorageLocation(jobId2, key);
+			}
+			assertTrue(blobFile.delete());
+			blobFile = server.getStorageLocation(jobId2, key);
+			assertTrue(blobFile.delete());
+			verifyDeleted(cache, jobId2, key, highAvailability);
+		}
+	}
+
+	@Test
+	public void testGetFailsIncomingNoJob() throws IOException {
+		testGetFailsIncoming(null, false);
+	}
+
+	@Test
+	public void testGetFailsIncomingForJob() throws IOException {
+		testGetFailsIncoming(new JobID(), false);
+	}
+
+	@Test
+	public void testGetFailsIncomingForJobHa() throws IOException {
+		testGetFailsIncoming(new JobID(), true);
+	}
+
+	/**
+	 * Retrieves a BLOB via a {@link BlobCache} which cannot create incoming files. File transfers
+	 * should fail.
+	 *
+	 * @param jobId
+	 * 		job id
+	 * @param highAvailability
+	 * 		whether to retrieve a permanent blob (<tt>true</tt>, via {@link
+	 * 		PermanentBlobCache#getHAFile(JobID, BlobKey)}) or not (<tt>false</tt>, via {@link
+	 * 		TransientBlobCache#getFile(JobID, BlobKey)})
+	 */
+	private void testGetFailsIncoming(@Nullable final JobID jobId, boolean highAvailability)
+			throws IOException {
+		assumeTrue(!OperatingSystem.isWindows()); //setWritable doesn't work on Windows.
+
+		final Configuration config = new Configuration();
+		config.setString(BlobServerOptions.STORAGE_DIRECTORY, temporaryFolder.newFolder().getAbsolutePath());
+
+		File tempFileDir = null;
+		try (
+			BlobServer server = new BlobServer(config, new VoidBlobStore());
+			BlobCache cache = new BlobCache(new InetSocketAddress("localhost", server.getPort()),
+				config, new VoidBlobStore())) {
+
+			server.start();
+
+			// store the data on the server
+			byte[] data = new byte[2000000];
+			rnd.nextBytes(data);
+			BlobKey blobKey = put(server, jobId, data, highAvailability);
+
+			// make sure the blob cache cannot create any files in its storage dir
+			if (highAvailability) {
+				tempFileDir =
+					cache.getPermanentBlobStore().createTemporaryFilename().getParentFile();
+			} else {
+				tempFileDir =
+					cache.getTransientBlobStore().createTemporaryFilename().getParentFile();
+			}
+			assertTrue(tempFileDir.setExecutable(true, false));
+			assertTrue(tempFileDir.setReadable(true, false));
+			assertTrue(tempFileDir.setWritable(false, false));
+
+			// request the file from the server via the cache
+			exception.expect(IOException.class);
+			exception.expectMessage("Failed to fetch BLOB ");
+
+			try {
+				get(cache, jobId, blobKey, highAvailability);
+			} finally {
+				HashSet<String> expectedDirs = new HashSet<>();
+				expectedDirs.add("incoming");
+				if (jobId != null) {
+					// only the incoming and job directory should exist (no job directory!)
+					expectedDirs.add(JOB_DIR_PREFIX + jobId);
+					File storageDir = tempFileDir.getParentFile();
+					String[] actualDirs = storageDir.list();
+					assertNotNull(actualDirs);
+					assertEquals(expectedDirs, new HashSet<>(Arrays.asList(actualDirs)));
+
+					// job directory should be empty
+					File jobDir = new File(tempFileDir.getParentFile(), JOB_DIR_PREFIX + jobId);
+					assertArrayEquals(new String[] {}, jobDir.list());
+				} else {
+					// only the incoming and no_job directory should exist (no job directory!)
+					expectedDirs.add(NO_JOB_DIR_PREFIX);
+					File storageDir = tempFileDir.getParentFile();
+					String[] actualDirs = storageDir.list();
+					assertNotNull(actualDirs);
+					assertEquals(expectedDirs, new HashSet<>(Arrays.asList(actualDirs)));
+
+					// no_job directory should be empty
+					File noJobDir = new File(tempFileDir.getParentFile(), NO_JOB_DIR_PREFIX);
+					assertArrayEquals(new String[] {}, noJobDir.list());
+				}
+			}
+		} finally {
+			// set writable again to make sure we can remove the directory
+			if (tempFileDir != null) {
+				//noinspection ResultOfMethodCallIgnored
+				tempFileDir.setWritable(true, false);
+			}
+		}
+	}
+
+	@Test
+	public void testGetFailsStoreNoJob() throws IOException {
+		testGetFailsStore(null, false);
+	}
+
+	@Test
+	public void testGetFailsStoreForJob() throws IOException {
+		testGetFailsStore(new JobID(), false);
+	}
+
+	@Test
+	public void testGetFailsStoreForJobHa() throws IOException {
+		testGetFailsStore(new JobID(), true);
+	}
+
+	/**
+	 * Retrieves a BLOB via a {@link BlobCache} which cannot create the final storage file. File
+	 * transfers should fail.
+	 *
+	 * @param jobId
+	 * 		job id
+	 * @param highAvailability
+	 * 		whether to retrieve a permanent blob (<tt>true</tt>, via {@link
+	 * 		PermanentBlobCache#getHAFile(JobID, BlobKey)}) or not (<tt>false</tt>, via {@link
+	 * 		TransientBlobCache#getFile(JobID, BlobKey)})
+	 */
+	private void testGetFailsStore(@Nullable final JobID jobId, boolean highAvailability)
+			throws IOException {
+		assumeTrue(!OperatingSystem.isWindows()); //setWritable doesn't work on Windows.
+
+		final Configuration config = new Configuration();
+		config.setString(BlobServerOptions.STORAGE_DIRECTORY, temporaryFolder.newFolder().getAbsolutePath());
+
+		File jobStoreDir = null;
+		try (
+			BlobServer server = new BlobServer(config, new VoidBlobStore());
+			BlobCache cache = new BlobCache(new InetSocketAddress("localhost", server.getPort()),
+				config, new VoidBlobStore())) {
+
+			server.start();
+
+			// store the data on the server
+			byte[] data = new byte[2000000];
+			rnd.nextBytes(data);
+			BlobKey blobKey = put(server, jobId, data, highAvailability);
+
+			// make sure the blob cache cannot create any files in its storage dir
+			if (highAvailability) {
+				jobStoreDir = cache.getPermanentBlobStore().getStorageLocation(jobId, new BlobKey())
+					.getParentFile();
+			} else {
+				jobStoreDir = cache.getTransientBlobStore().getStorageLocation(jobId, new BlobKey())
+					.getParentFile();
+			}
+			assertTrue(jobStoreDir.setExecutable(true, false));
+			assertTrue(jobStoreDir.setReadable(true, false));
+			assertTrue(jobStoreDir.setWritable(false, false));
+
+			// request the file from the server via the cache
+			exception.expect(AccessDeniedException.class);
+
+			try {
+				get(cache, jobId, blobKey, highAvailability);
+			} finally {
+				// there should be no remaining incoming files
+				File incomingFileDir = new File(jobStoreDir.getParent(), "incoming");
+				assertArrayEquals(new String[] {}, incomingFileDir.list());
+
+				// there should be no files in the job directory
+				assertArrayEquals(new String[] {}, jobStoreDir.list());
+			}
+		} finally {
+			// set writable again to make sure we can remove the directory
+			if (jobStoreDir != null) {
+				//noinspection ResultOfMethodCallIgnored
+				jobStoreDir.setWritable(true, false);
+			}
+		}
+	}
+
+	/**
+	 * Retrieves a BLOB from the HA store to a {@link BlobServer} whose HA store does not contain
+	 * the file. File transfers should fail.
+	 */
+	@Test
+	public void testGetFailsHaStoreForJobHa() throws IOException {
+		final JobID jobId = new JobID();
+
+		final Configuration config = new Configuration();
+		config.setString(BlobServerOptions.STORAGE_DIRECTORY, temporaryFolder.newFolder().getAbsolutePath());
+
+		try (
+			BlobServer server = new BlobServer(config, new VoidBlobStore());
+			BlobCache cache = new BlobCache(new InetSocketAddress("localhost", server.getPort()),
+				config, new VoidBlobStore())) {
+
+			server.start();
+
+			// store the data on the server (and blobStore), remove from local server store
+			byte[] data = new byte[2000000];
+			rnd.nextBytes(data);
+			BlobKey blobKey = put(server, jobId, data, true);
+			assertTrue(server.getStorageLocation(jobId, blobKey).delete());
+
+			File tempFileDir = server.createTemporaryFilename().getParentFile();
+
+			// request the file from the server via the cache
+			exception.expect(IOException.class);
+			exception.expectMessage("Failed to fetch BLOB ");
+
+			try {
+				get(cache, jobId, blobKey, true);
+			} finally {
+				HashSet<String> expectedDirs = new HashSet<>();
+				expectedDirs.add("incoming");
+				expectedDirs.add(JOB_DIR_PREFIX + jobId);
+				// only the incoming and job directory should exist (no job directory!)
+				File storageDir = tempFileDir.getParentFile();
+				String[] actualDirs = storageDir.list();
+				assertNotNull(actualDirs);
+				assertEquals(expectedDirs, new HashSet<>(Arrays.asList(actualDirs)));
+
+				// job directory should be empty
+				File jobDir = new File(tempFileDir.getParentFile(), JOB_DIR_PREFIX + jobId);
+				assertArrayEquals(new String[] {}, jobDir.list());
+			}
+		}
+	}
+
+	/**
+	 * FLINK-6020
+	 *
+	 * <p>Tests that concurrent get operations don't concurrently access the BlobStore to download a blob.
+	 */
+	@Test
+	public void testConcurrentGetOperationsNoJob() throws IOException, ExecutionException, InterruptedException {
+		testConcurrentGetOperations(null, false, false);
+	}
+
+	@Test
+	public void testConcurrentGetOperationsForJob() throws IOException, ExecutionException, InterruptedException {
+		testConcurrentGetOperations(new JobID(), false, false);
+	}
+
+	@Test
+	public void testConcurrentGetOperationsForJobHa() throws IOException, ExecutionException, InterruptedException {
+		testConcurrentGetOperations(new JobID(), true, false);
+	}
+
+	@Test
+	public void testConcurrentGetOperationsForJobHa2() throws IOException, ExecutionException, InterruptedException {
+		testConcurrentGetOperations(new JobID(), true, true);
+	}
+
+	/**
+	 * [FLINK-6020] Tests that concurrent get operations don't concurrently access the BlobStore to
+	 * download a blob.
+	 *
+	 * @param jobId
+	 * 		job ID to use (or <tt>null</tt> if job-unrelated)
+	 * @param highAvailability
+	 * 		whether to use permanent (<tt>true</tt>) or transient BLOBs (<tt>false</tt>)
+	 * @param cacheAccessesHAStore
+	 * 		whether the cache has access to the {@link BlobServer}'s HA store or not
+	 */
+	private void testConcurrentGetOperations(final JobID jobId, final boolean highAvailability,
+			final boolean cacheAccessesHAStore)
+			throws IOException, InterruptedException, ExecutionException {
+		final Configuration config = new Configuration();
+		config.setString(BlobServerOptions.STORAGE_DIRECTORY, temporaryFolder.newFolder().getAbsolutePath());
+
+		final BlobStore blobStoreServer = mock(BlobStore.class);
+		final BlobStore blobStoreCache = mock(BlobStore.class);
+
+		final int numberConcurrentGetOperations = 3;
+		final List<CompletableFuture<File>> getOperations = new ArrayList<>(numberConcurrentGetOperations);
+
+		final byte[] data = {1, 2, 3, 4, 99, 42};
+
+		MessageDigest md = BlobUtils.createMessageDigest();
+
+		// create the correct blob key by hashing our input data
+		final BlobKey blobKey = new BlobKey(md.digest(data));
+
+		final ExecutorService executor = Executors.newFixedThreadPool(numberConcurrentGetOperations);
+
+		try (
+			final BlobServer server = new BlobServer(config, blobStoreServer);
+			final BlobCache cache = new BlobCache(new InetSocketAddress("localhost", server.getPort()),
+				config, cacheAccessesHAStore ? blobStoreServer : blobStoreCache)) {
+
+			server.start();
+
+			// upload data first
+			assertEquals(blobKey, put(server, jobId, data, highAvailability));
+
+			// now try accessing it concurrently (only HA mode will be able to retrieve it from HA store!)
+			for (int i = 0; i < numberConcurrentGetOperations; i++) {
+				CompletableFuture<File> getOperation = CompletableFuture
+					.supplyAsync(
+						() -> {
+							try {
+								File file = get(cache, jobId, blobKey, highAvailability);
+								// check that we have read the right data
+								validateGetAndClose(new FileInputStream(file), data);
+								return file;
+							} catch (IOException e) {
+								throw new CompletionException(new FlinkException(
+									"Could not upload blob.", e));
+							}
+						}, executor);
+
+				getOperations.add(getOperation);
+			}
+
+			FutureUtils.ConjunctFuture<Collection<File>> filesFuture = FutureUtils.combineAll(getOperations);
+
+			// wait until all operations have completed and check that no exception was thrown
+			filesFuture.get();
+		} finally {
+			executor.shutdownNow();
+		}
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/071e27f7/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCachePutTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCachePutTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCachePutTest.java
new file mode 100644
index 0000000..7066b95
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCachePutTest.java
@@ -0,0 +1,808 @@
+/*
+ * 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.blob;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.configuration.BlobServerOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.core.testutils.CheckedThread;
+import org.apache.flink.runtime.concurrent.FutureUtils;
+import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.OperatingSystem;
+import org.apache.flink.util.TestLogger;
+
+import org.apache.commons.io.FileUtils;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.rules.TemporaryFolder;
+
+import javax.annotation.Nullable;
+
+import java.io.ByteArrayInputStream;
+import java.io.File;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.function.Supplier;
+
+import static org.apache.flink.runtime.blob.BlobServerPutTest.BlockingInputStream;
+import static org.apache.flink.runtime.blob.BlobServerPutTest.ChunkedInputStream;
+import static org.apache.flink.runtime.blob.BlobServerPutTest.put;
+import static org.apache.flink.runtime.blob.BlobServerPutTest.verifyContents;
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assume.assumeTrue;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+/**
+ * Tests for successful and failing PUT operations against the BLOB server,
+ * and successful GET operations.
+ */
+public class BlobCachePutTest extends TestLogger {
+
+	private final Random rnd = new Random();
+
+	@Rule
+	public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+	@Rule
+	public final ExpectedException exception = ExpectedException.none();
+
+	// --- concurrency tests for utility methods which could fail during the put operation ---
+
+	/**
+	 * Checked thread that calls {@link TransientBlobCache#getStorageLocation(JobID, BlobKey)}.
+	 */
+	public static class TransientBlobCacheGetStorageLocation extends CheckedThread {
+		private final TransientBlobCache cache;
+		private final JobID jobId;
+		private final BlobKey key;
+
+		TransientBlobCacheGetStorageLocation(
+				TransientBlobCache cache, @Nullable JobID jobId, BlobKey key) {
+			this.cache = cache;
+			this.jobId = jobId;
+			this.key = key;
+		}
+
+		@Override
+		public void go() throws Exception {
+			cache.getStorageLocation(jobId, key);
+		}
+	}
+
+	/**
+	 * Checked thread that calls {@link PermanentBlobCache#getStorageLocation(JobID, BlobKey)}.
+	 */
+	public static class PermanentBlobCacheGetStorageLocation extends CheckedThread {
+		private final PermanentBlobCache cache;
+		private final JobID jobId;
+		private final BlobKey key;
+
+		PermanentBlobCacheGetStorageLocation(PermanentBlobCache cache, JobID jobId, BlobKey key) {
+			this.cache = cache;
+			this.jobId = jobId;
+			this.key = key;
+		}
+
+		@Override
+		public void go() throws Exception {
+			cache.getStorageLocation(jobId, key);
+		}
+	}
+
+	/**
+	 * Tests concurrent calls to {@link TransientBlobCache#getStorageLocation(JobID, BlobKey)}.
+	 */
+	@Test
+	public void testTransientBlobCacheGetStorageLocationConcurrentNoJob() throws Exception {
+		testTransientBlobCacheGetStorageLocationConcurrent(null);
+	}
+
+	/**
+	 * Tests concurrent calls to {@link TransientBlobCache#getStorageLocation(JobID, BlobKey)}.
+	 */
+	@Test
+	public void testTransientBlobCacheGetStorageLocationConcurrentForJob() throws Exception {
+		testTransientBlobCacheGetStorageLocationConcurrent(new JobID());
+	}
+
+	private void testTransientBlobCacheGetStorageLocationConcurrent(
+			@Nullable final JobID jobId) throws Exception {
+		final Configuration config = new Configuration();
+		config.setString(BlobServerOptions.STORAGE_DIRECTORY, temporaryFolder.newFolder().getAbsolutePath());
+
+		try (BlobServer server = new BlobServer(config, new VoidBlobStore());
+			final TransientBlobCache cache = new TransientBlobCache(
+				new InetSocketAddress("localhost", server.getPort()), config)) {
+
+			server.start();
+
+			BlobKey key = new BlobKey();
+			CheckedThread[] threads = new CheckedThread[] {
+				new TransientBlobCacheGetStorageLocation(cache, jobId, key),
+				new TransientBlobCacheGetStorageLocation(cache, jobId, key),
+				new TransientBlobCacheGetStorageLocation(cache, jobId, key)
+			};
+			checkedThreadSimpleTest(threads);
+		}
+	}
+
+	/**
+	 * Tests concurrent calls to {@link PermanentBlobCache#getStorageLocation(JobID, BlobKey)}.
+	 */
+	@Test
+	public void testPermanentBlobCacheGetStorageLocationConcurrentForJob() throws Exception {
+		final JobID jobId = new JobID();
+		final Configuration config = new Configuration();
+		config.setString(BlobServerOptions.STORAGE_DIRECTORY, temporaryFolder.newFolder().getAbsolutePath());
+
+		try (BlobServer server = new BlobServer(config, new VoidBlobStore());
+			final PermanentBlobCache cache = new PermanentBlobCache(
+				new InetSocketAddress("localhost", server.getPort()), config,
+				new VoidBlobStore())) {
+
+			server.start();
+
+			BlobKey key = new BlobKey();
+			CheckedThread[] threads = new CheckedThread[] {
+				new PermanentBlobCacheGetStorageLocation(cache, jobId, key),
+				new PermanentBlobCacheGetStorageLocation(cache, jobId, key),
+				new PermanentBlobCacheGetStorageLocation(cache, jobId, key)
+			};
+			checkedThreadSimpleTest(threads);
+		}
+	}
+
+	/**
+	 * Helper method to first start all threads and then wait for their completion.
+	 *
+	 * @param threads threads to use
+	 * @throws Exception exceptions that are thrown from the threads
+	 */
+	private void checkedThreadSimpleTest(CheckedThread[] threads)
+		throws Exception {
+
+		// start all threads
+		for (CheckedThread t: threads) {
+			t.start();
+		}
+
+		// wait for thread completion and check exceptions
+		for (CheckedThread t: threads) {
+			t.sync();
+		}
+	}
+
+	// --------------------------------------------------------------------------------------------
+
+	@Test
+	public void testPutBufferSuccessfulGet1() throws IOException {
+		testPutBufferSuccessfulGet(null, null, false);
+	}
+
+	@Test
+	public void testPutBufferSuccessfulGet2() throws IOException {
+		testPutBufferSuccessfulGet(null, new JobID(), false);
+	}
+
+	@Test
+	public void testPutBufferSuccessfulGet3() throws IOException {
+		testPutBufferSuccessfulGet(new JobID(), new JobID(), false);
+	}
+
+	@Test
+	public void testPutBufferSuccessfulGet4() throws IOException {
+		testPutBufferSuccessfulGet(new JobID(), null, false);
+	}
+
+	@Test
+	public void testPutBufferSuccessfulGetHa() throws IOException {
+		testPutBufferSuccessfulGet(new JobID(), new JobID(), true);
+	}
+
+	/**
+	 * Uploads two byte arrays for different jobs into the server via the {@link BlobCache}. File
+	 * transfers should be successful.
+	 *
+	 * @param jobId1
+	 * 		first job id
+	 * @param jobId2
+	 * 		second job id
+	 * @param highAvailability
+	 * 		whether to upload a permanent blob (<tt>true</tt>, via {@link
+	 * 		BlobClient#uploadJarFiles(InetSocketAddress, Configuration, JobID, List)}) or not
+	 * 		(<tt>false</tt>, via {@link TransientBlobCache#put(JobID, byte[])}
+	 */
+	private void testPutBufferSuccessfulGet(
+			@Nullable JobID jobId1, @Nullable JobID jobId2, boolean highAvailability)
+			throws IOException {
+
+		final Configuration config = new Configuration();
+		config.setString(BlobServerOptions.STORAGE_DIRECTORY, temporaryFolder.newFolder().getAbsolutePath());
+
+		try (
+			BlobServer server = new BlobServer(config, new VoidBlobStore());
+			BlobCache cache = new BlobCache(new InetSocketAddress("localhost", server.getPort()),
+				config, new VoidBlobStore())) {
+
+			server.start();
+
+			byte[] data = new byte[2000000];
+			rnd.nextBytes(data);
+			byte[] data2 = Arrays.copyOfRange(data, 10, 54);
+
+			// put data for jobId1 and verify
+			BlobKey key1a = put(cache, jobId1, data, highAvailability);
+			assertNotNull(key1a);
+
+			BlobKey key1b = put(cache, jobId1, data2, highAvailability);
+			assertNotNull(key1b);
+
+			// files should be available on the server
+			verifyContents(server, jobId1, key1a, data, highAvailability);
+			verifyContents(server, jobId1, key1b, data2, highAvailability);
+
+			// now put data for jobId2 and verify that both are ok
+			BlobKey key2a = put(cache, jobId2, data, highAvailability);
+			assertNotNull(key2a);
+			assertEquals(key1a, key2a);
+
+			BlobKey key2b = put(cache, jobId2, data2, highAvailability);
+			assertNotNull(key2b);
+			assertEquals(key1b, key2b);
+
+			// verify the accessibility and the BLOB contents
+			verifyContents(server, jobId1, key1a, data, highAvailability);
+			verifyContents(server, jobId1, key1b, data2, highAvailability);
+			verifyContents(server, jobId2, key2a, data, highAvailability);
+			verifyContents(server, jobId2, key2b, data2, highAvailability);
+		}
+	}
+
+	// --------------------------------------------------------------------------------------------
+
+	@Test
+	public void testPutStreamSuccessfulGet1() throws IOException {
+		testPutStreamSuccessfulGet(null, null);
+	}
+
+	@Test
+	public void testPutStreamSuccessfulGet2() throws IOException {
+		testPutStreamSuccessfulGet(null, new JobID());
+	}
+
+	@Test
+	public void testPutStreamSuccessfulGet3() throws IOException {
+		testPutStreamSuccessfulGet(new JobID(), new JobID());
+	}
+
+	@Test
+	public void testPutStreamSuccessfulGet4() throws IOException {
+		testPutStreamSuccessfulGet(new JobID(), null);
+	}
+
+	/**
+	 * Uploads two file streams for different jobs into the server via the {@link BlobCache}. File
+	 * transfers should be successful.
+	 *
+	 * <p>Note that high-availability uploads of streams is currently only possible at the {@link
+	 * BlobServer}.
+	 *
+	 * @param jobId1
+	 * 		first job id
+	 * @param jobId2
+	 * 		second job id
+	 */
+	private void testPutStreamSuccessfulGet(@Nullable JobID jobId1, @Nullable JobID jobId2)
+			throws IOException {
+
+		final Configuration config = new Configuration();
+		config.setString(BlobServerOptions.STORAGE_DIRECTORY, temporaryFolder.newFolder().getAbsolutePath());
+
+		try (
+			BlobServer server = new BlobServer(config, new VoidBlobStore());
+			BlobCache cache = new BlobCache(new InetSocketAddress("localhost", server.getPort()),
+				config, new VoidBlobStore())) {
+
+			server.start();
+
+			byte[] data = new byte[2000000];
+			rnd.nextBytes(data);
+			byte[] data2 = Arrays.copyOfRange(data, 10, 54);
+
+			// put data for jobId1 and verify
+			BlobKey key1a = put(cache, jobId1, new ByteArrayInputStream(data), false);
+			assertNotNull(key1a);
+
+			BlobKey key1b = put(cache, jobId1, new ByteArrayInputStream(data2), false);
+			assertNotNull(key1b);
+
+			// files should be available on the server
+			verifyContents(server, jobId1, key1a, data, false);
+			verifyContents(server, jobId1, key1b, data2, false);
+
+			// now put data for jobId2 and verify that both are ok
+			BlobKey key2a = put(cache, jobId2, new ByteArrayInputStream(data), false);
+			assertNotNull(key2a);
+			assertEquals(key1a, key2a);
+
+			BlobKey key2b = put(cache, jobId2, new ByteArrayInputStream(data2), false);
+			assertNotNull(key2b);
+			assertEquals(key1b, key2b);
+
+			// verify the accessibility and the BLOB contents
+			verifyContents(server, jobId1, key1a, data, false);
+			verifyContents(server, jobId1, key1b, data2, false);
+			verifyContents(server, jobId2, key2a, data, false);
+			verifyContents(server, jobId2, key2b, data2, false);
+		}
+	}
+
+	// --------------------------------------------------------------------------------------------
+
+	@Test
+	public void testPutChunkedStreamSuccessfulGet1() throws IOException {
+		testPutChunkedStreamSuccessfulGet(null, null);
+	}
+
+	@Test
+	public void testPutChunkedStreamSuccessfulGet2() throws IOException {
+		testPutChunkedStreamSuccessfulGet(null, new JobID());
+	}
+
+	@Test
+	public void testPutChunkedStreamSuccessfulGet3() throws IOException {
+		testPutChunkedStreamSuccessfulGet(new JobID(), new JobID());
+	}
+
+	@Test
+	public void testPutChunkedStreamSuccessfulGet4() throws IOException {
+		testPutChunkedStreamSuccessfulGet(new JobID(), null);
+	}
+
+	/**
+	 * Uploads two chunked file streams for different jobs into the server via the {@link
+	 * BlobCache}. File transfers should be successful.
+	 *
+	 * <p>Note that high-availability uploads of streams is currently only possible at the {@link
+	 * BlobServer}.
+	 *
+	 * @param jobId1
+	 * 		first job id
+	 * @param jobId2
+	 * 		second job id
+	 */
+	private void testPutChunkedStreamSuccessfulGet(
+			@Nullable JobID jobId1, @Nullable JobID jobId2) throws IOException {
+
+		final Configuration config = new Configuration();
+		config.setString(BlobServerOptions.STORAGE_DIRECTORY, temporaryFolder.newFolder().getAbsolutePath());
+
+		try (
+			BlobServer server = new BlobServer(config, new VoidBlobStore());
+			BlobCache cache = new BlobCache(new InetSocketAddress("localhost", server.getPort()),
+				config, new VoidBlobStore())) {
+
+			server.start();
+
+			byte[] data = new byte[2000000];
+			rnd.nextBytes(data);
+			byte[] data2 = Arrays.copyOfRange(data, 10, 54);
+
+			// put data for jobId1 and verify
+			BlobKey key1a = put(cache, jobId1, new ChunkedInputStream(data, 19), false);
+			assertNotNull(key1a);
+
+			BlobKey key1b = put(cache, jobId1, new ChunkedInputStream(data2, 19), false);
+			assertNotNull(key1b);
+
+			// files should be available on the server
+			verifyContents(server, jobId1, key1a, data, false);
+			verifyContents(server, jobId1, key1b, data2, false);
+
+			// now put data for jobId2 and verify that both are ok
+			BlobKey key2a = put(cache, jobId2, new ChunkedInputStream(data, 19), false);
+			assertNotNull(key2a);
+			assertEquals(key1a, key2a);
+
+			BlobKey key2b = put(cache, jobId2, new ChunkedInputStream(data2, 19), false);
+			assertNotNull(key2b);
+			assertEquals(key1b, key2b);
+
+			// verify the accessibility and the BLOB contents
+			verifyContents(server, jobId1, key1a, data, false);
+			verifyContents(server, jobId1, key1b, data2, false);
+			verifyContents(server, jobId2, key2a, data, false);
+			verifyContents(server, jobId2, key2b, data2, false);
+		}
+	}
+
+	// --------------------------------------------------------------------------------------------
+
+	@Test
+	public void testPutBufferFailsNoJob() throws IOException {
+		testPutBufferFails(null, false);
+	}
+
+	@Test
+	public void testPutBufferFailsForJob() throws IOException {
+		testPutBufferFails(new JobID(), false);
+	}
+
+	@Test
+	public void testPutBufferFailsForJobHa() throws IOException {
+		testPutBufferFails(new JobID(), true);
+	}
+
+	/**
+	 * Uploads a byte array to a server which cannot create any files via the {@link BlobCache}.
+	 * File transfers should fail.
+	 *
+	 * @param jobId
+	 * 		job id
+	 * @param highAvailability
+	 * 		whether to upload a permanent blob (<tt>true</tt>, via {@link
+	 * 		BlobClient#uploadJarFiles(InetSocketAddress, Configuration, JobID, List)}) or not
+	 * 		(<tt>false</tt>, via {@link TransientBlobCache#put(JobID, byte[])}
+	 */
+	private void testPutBufferFails(@Nullable final JobID jobId, boolean highAvailability)
+			throws IOException {
+		assumeTrue(!OperatingSystem.isWindows()); //setWritable doesn't work on Windows.
+
+		final Configuration config = new Configuration();
+		config.setString(BlobServerOptions.STORAGE_DIRECTORY, temporaryFolder.newFolder().getAbsolutePath());
+
+		File tempFileDir = null;
+		try (
+			BlobServer server = new BlobServer(config, new VoidBlobStore());
+			BlobCache cache = new BlobCache(new InetSocketAddress("localhost", server.getPort()),
+				config, new VoidBlobStore())) {
+
+			server.start();
+
+			// make sure the blob server cannot create any files in its storage dir
+			tempFileDir = server.createTemporaryFilename().getParentFile().getParentFile();
+			assertTrue(tempFileDir.setExecutable(true, false));
+			assertTrue(tempFileDir.setReadable(true, false));
+			assertTrue(tempFileDir.setWritable(false, false));
+
+			byte[] data = new byte[2000000];
+			rnd.nextBytes(data);
+
+			// upload the file to the server via the cache
+			exception.expect(IOException.class);
+			exception.expectMessage("PUT operation failed: ");
+
+			put(cache, jobId, data, highAvailability);
+
+		} finally {
+			// set writable again to make sure we can remove the directory
+			if (tempFileDir != null) {
+				//noinspection ResultOfMethodCallIgnored
+				tempFileDir.setWritable(true, false);
+			}
+		}
+	}
+
+	@Test
+	public void testPutBufferFailsIncomingNoJob() throws IOException {
+		testPutBufferFailsIncoming(null, false);
+	}
+
+	@Test
+	public void testPutBufferFailsIncomingForJob() throws IOException {
+		testPutBufferFailsIncoming(new JobID(), false);
+	}
+
+	@Test
+	public void testPutBufferFailsIncomingForJobHa() throws IOException {
+		testPutBufferFailsIncoming(new JobID(), true);
+	}
+
+	/**
+	 * Uploads a byte array to a server which cannot create incoming files via the {@link
+	 * BlobCache}. File transfers should fail.
+	 *
+	 * @param jobId
+	 * 		job id
+	 * @param highAvailability
+	 * 		whether to upload a permanent blob (<tt>true</tt>, via {@link
+	 * 		BlobClient#uploadJarFiles(InetSocketAddress, Configuration, JobID, List)}) or not
+	 * 		(<tt>false</tt>, via {@link TransientBlobCache#put(JobID, byte[])}
+	 */
+	private void testPutBufferFailsIncoming(@Nullable final JobID jobId, boolean highAvailability)
+			throws IOException {
+		assumeTrue(!OperatingSystem.isWindows()); //setWritable doesn't work on Windows.
+
+		final Configuration config = new Configuration();
+		config.setString(BlobServerOptions.STORAGE_DIRECTORY, temporaryFolder.newFolder().getAbsolutePath());
+
+		File tempFileDir = null;
+		try (
+			BlobServer server = new BlobServer(config, new VoidBlobStore());
+			BlobCache cache = new BlobCache(new InetSocketAddress("localhost", server.getPort()),
+				config, new VoidBlobStore())) {
+
+			server.start();
+
+			// make sure the blob server cannot create any files in its storage dir
+			tempFileDir = server.createTemporaryFilename().getParentFile();
+			assertTrue(tempFileDir.setExecutable(true, false));
+			assertTrue(tempFileDir.setReadable(true, false));
+			assertTrue(tempFileDir.setWritable(false, false));
+
+			byte[] data = new byte[2000000];
+			rnd.nextBytes(data);
+
+			// upload the file to the server via the cache
+			exception.expect(IOException.class);
+			exception.expectMessage("PUT operation failed: ");
+
+			try {
+				put(cache, jobId, data, highAvailability);
+			} finally {
+				File storageDir = tempFileDir.getParentFile();
+				// only the incoming directory should exist (no job directory!)
+				assertArrayEquals(new String[] {"incoming"}, storageDir.list());
+			}
+		} finally {
+			// set writable again to make sure we can remove the directory
+			if (tempFileDir != null) {
+				//noinspection ResultOfMethodCallIgnored
+				tempFileDir.setWritable(true, false);
+			}
+		}
+	}
+
+	@Test
+	public void testPutBufferFailsStoreNoJob() throws IOException {
+		testPutBufferFailsStore(null, false);
+	}
+
+	@Test
+	public void testPutBufferFailsStoreForJob() throws IOException {
+		testPutBufferFailsStore(new JobID(), false);
+	}
+
+	@Test
+	public void testPutBufferFailsStoreForJobHa() throws IOException {
+		testPutBufferFailsStore(new JobID(), true);
+	}
+
+	/**
+	 * Uploads a byte array to a server which cannot create files via the {@link BlobCache}. File
+	 * transfers should fail.
+	 *
+	 * @param jobId
+	 * 		job id
+	 * @param highAvailability
+	 * 		whether to upload a permanent blob (<tt>true</tt>, via {@link
+	 * 		BlobClient#uploadJarFiles(InetSocketAddress, Configuration, JobID, List)}) or not
+	 * 		(<tt>false</tt>, via {@link TransientBlobCache#put(JobID, byte[])}
+	 */
+	private void testPutBufferFailsStore(@Nullable final JobID jobId, boolean highAvailability)
+			throws IOException {
+		assumeTrue(!OperatingSystem.isWindows()); //setWritable doesn't work on Windows.
+
+		final Configuration config = new Configuration();
+		config.setString(BlobServerOptions.STORAGE_DIRECTORY, temporaryFolder.newFolder().getAbsolutePath());
+
+		File jobStoreDir = null;
+		try (
+			BlobServer server = new BlobServer(config, new VoidBlobStore());
+			BlobCache cache = new BlobCache(new InetSocketAddress("localhost", server.getPort()),
+				config, new VoidBlobStore())) {
+
+			server.start();
+
+			// make sure the blob server cannot create any files in its storage dir
+			jobStoreDir = server.getStorageLocation(jobId, new BlobKey()).getParentFile();
+			assertTrue(jobStoreDir.setExecutable(true, false));
+			assertTrue(jobStoreDir.setReadable(true, false));
+			assertTrue(jobStoreDir.setWritable(false, false));
+
+			byte[] data = new byte[2000000];
+			rnd.nextBytes(data);
+
+			// upload the file to the server via the cache
+			exception.expect(IOException.class);
+			exception.expectMessage("PUT operation failed: ");
+
+			try {
+				put(cache, jobId, data, highAvailability);
+			} finally {
+				// there should be no remaining incoming files
+				File incomingFileDir = new File(jobStoreDir.getParent(), "incoming");
+				assertArrayEquals(new String[] {}, incomingFileDir.list());
+
+				// there should be no files in the job directory
+				assertArrayEquals(new String[] {}, jobStoreDir.list());
+			}
+		} finally {
+			// set writable again to make sure we can remove the directory
+			if (jobStoreDir != null) {
+				//noinspection ResultOfMethodCallIgnored
+				jobStoreDir.setWritable(true, false);
+			}
+		}
+	}
+
+	@Test
+	public void testConcurrentPutOperationsNoJob() throws IOException, ExecutionException, InterruptedException {
+		testConcurrentPutOperations(null, false);
+	}
+
+	@Test
+	public void testConcurrentPutOperationsForJob() throws IOException, ExecutionException, InterruptedException {
+		testConcurrentPutOperations(new JobID(), false);
+	}
+
+	@Test
+	public void testConcurrentPutOperationsForJobHa() throws IOException, ExecutionException, InterruptedException {
+		testConcurrentPutOperations(new JobID(), true);
+	}
+
+	/**
+	 * [FLINK-6020]
+	 * Tests that concurrent put operations will only upload the file once to the {@link BlobStore}
+	 * and that the files are not corrupt at any time.
+	 *
+	 * @param jobId
+	 * 		job ID to use (or <tt>null</tt> if job-unrelated)
+	 * @param highAvailability
+	 * 		whether to use permanent (<tt>true</tt>) or transient BLOBs (<tt>false</tt>)
+	 */
+	private void testConcurrentPutOperations(
+			@Nullable final JobID jobId, final boolean highAvailability)
+			throws IOException, InterruptedException, ExecutionException {
+		final Configuration config = new Configuration();
+		config.setString(BlobServerOptions.STORAGE_DIRECTORY, temporaryFolder.newFolder().getAbsolutePath());
+
+		final BlobStore blobStoreServer = mock(BlobStore.class);
+		final BlobStore blobStoreCache = mock(BlobStore.class);
+
+		int concurrentPutOperations = 2;
+		int dataSize = 1024;
+
+		final CountDownLatch countDownLatch = new CountDownLatch(concurrentPutOperations);
+		final byte[] data = new byte[dataSize];
+
+		final List<Path> jars;
+		if (highAvailability) {
+			// implement via JAR file upload instead:
+			File tmpFile = temporaryFolder.newFile();
+			FileUtils.writeByteArrayToFile(tmpFile, data);
+			jars = Collections.singletonList(new Path(tmpFile.getAbsolutePath()));
+		} else {
+			jars = null;
+		}
+
+		Collection<CompletableFuture<BlobKey>> allFutures = new ArrayList<>(concurrentPutOperations);
+
+		ExecutorService executor = Executors.newFixedThreadPool(concurrentPutOperations);
+
+		try (
+			final BlobServer server = new BlobServer(config, blobStoreServer);
+			final BlobCache cache = new BlobCache(new InetSocketAddress("localhost", server.getPort()),
+				config, blobStoreCache)) {
+
+			server.start();
+
+			// for highAvailability
+			final InetSocketAddress serverAddress =
+				new InetSocketAddress("localhost", server.getPort());
+			// uploading HA BLOBs works on BlobServer only (and, for now, via the BlobClient)
+
+			for (int i = 0; i < concurrentPutOperations; i++) {
+				final Supplier<BlobKey> callable;
+				if (highAvailability) {
+					// cannot use a blocking stream here (upload only possible via files)
+					callable =
+						() -> {
+							try {
+								List<BlobKey> keys =
+									BlobClient.uploadJarFiles(serverAddress, config, jobId, jars);
+								assertEquals(1, keys.size());
+								BlobKey uploadedKey = keys.get(0);
+								// check the uploaded file's contents (concurrently)
+								verifyContents(server, jobId, uploadedKey, data, true);
+								return uploadedKey;
+							} catch (IOException e) {
+								throw new CompletionException(new FlinkException(
+									"Could not upload blob.", e));
+							}
+						};
+
+				} else {
+					callable =
+						() -> {
+							try {
+								BlockingInputStream inputStream =
+									new BlockingInputStream(countDownLatch, data);
+								BlobKey uploadedKey = put(cache, jobId, inputStream, false);
+								// check the uploaded file's contents (concurrently)
+								verifyContents(server, jobId, uploadedKey, data, false);
+								return uploadedKey;
+							} catch (IOException e) {
+								throw new CompletionException(new FlinkException(
+									"Could not upload blob.", e));
+							}
+					};
+				}
+				CompletableFuture<BlobKey> putFuture = CompletableFuture
+					.supplyAsync(callable, executor);
+
+				allFutures.add(putFuture);
+			}
+
+			FutureUtils.ConjunctFuture<Collection<BlobKey>> conjunctFuture = FutureUtils.combineAll(allFutures);
+
+			// wait until all operations have completed and check that no exception was thrown
+			Collection<BlobKey> blobKeys = conjunctFuture.get();
+
+			Iterator<BlobKey> blobKeyIterator = blobKeys.iterator();
+
+			assertTrue(blobKeyIterator.hasNext());
+
+			BlobKey blobKey = blobKeyIterator.next();
+
+			// make sure that all blob keys are the same
+			while (blobKeyIterator.hasNext()) {
+				assertEquals(blobKey, blobKeyIterator.next());
+			}
+
+			// check the uploaded file's contents
+			verifyContents(server, jobId, blobKey, data, highAvailability);
+
+			// check that we only uploaded the file once to the blob store
+			if (highAvailability) {
+				verify(blobStoreServer, times(1)).put(any(File.class), eq(jobId), eq(blobKey));
+			} else {
+				// can't really verify much in the other cases other than that the put operations should
+				// work and not corrupt files
+				verify(blobStoreServer, times(0)).put(any(File.class), eq(jobId), eq(blobKey));
+			}
+			// caches must not access the blob store (they are not allowed to write there)
+			verify(blobStoreCache, times(0)).put(any(File.class), eq(jobId), eq(blobKey));
+		} finally {
+			executor.shutdownNow();
+		}
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/071e27f7/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheRecoveryTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheRecoveryTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheRecoveryTest.java
new file mode 100644
index 0000000..9f462c2
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheRecoveryTest.java
@@ -0,0 +1,148 @@
+/*
+ * 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.blob;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.configuration.BlobServerOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.configuration.HighAvailabilityOptions;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.jobmanager.HighAvailabilityMode;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.Arrays;
+import java.util.Random;
+
+import static org.apache.flink.runtime.blob.BlobServerGetTest.verifyDeleted;
+import static org.apache.flink.runtime.blob.BlobServerPutTest.put;
+import static org.apache.flink.runtime.blob.BlobServerPutTest.verifyContents;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Tests for the recovery of files of a {@link BlobCache} from a HA store.
+ */
+public class BlobCacheRecoveryTest extends TestLogger {
+
+	@Rule
+	public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+	/**
+	 * Tests that with {@link HighAvailabilityMode#ZOOKEEPER} distributed JARs are recoverable from any
+	 * participating BlobServer.
+	 */
+	@Test
+	public void testBlobCacheRecovery() throws Exception {
+		Configuration config = new Configuration();
+		config.setString(HighAvailabilityOptions.HA_MODE, "ZOOKEEPER");
+		config.setString(CoreOptions.STATE_BACKEND, "FILESYSTEM");
+		config.setString(BlobServerOptions.STORAGE_DIRECTORY, temporaryFolder.newFolder().getAbsolutePath());
+		config.setString(HighAvailabilityOptions.HA_STORAGE_PATH, temporaryFolder.newFolder().getPath());
+
+		BlobStoreService blobStoreService = null;
+
+		try {
+			blobStoreService = BlobUtils.createBlobStoreFromConfig(config);
+
+			testBlobCacheRecovery(config, blobStoreService);
+		} finally {
+			if (blobStoreService != null) {
+				blobStoreService.closeAndCleanupAllData();
+			}
+		}
+	}
+
+	/**
+	 * Helper to test that the {@link BlobServer} recovery from its HA store works.
+	 *
+	 * <p>Uploads two BLOBs to one {@link BlobServer} via a {@link BlobCache} and expects a second
+	 * {@link BlobCache} to be able to retrieve them from a second {@link BlobServer} that is
+	 * configured with the same HA store.
+	 *
+	 * @param config
+	 * 		blob server configuration (including HA settings like {@link HighAvailabilityOptions#HA_STORAGE_PATH}
+	 * 		and {@link HighAvailabilityOptions#HA_CLUSTER_ID}) used to set up <tt>blobStore</tt>
+	 * @param blobStore
+	 * 		shared HA blob store to use
+	 *
+	 * @throws IOException
+	 * 		in case of failures
+	 */
+	public static void testBlobCacheRecovery(
+			final Configuration config, final BlobStore blobStore) throws IOException {
+
+		final String clusterId = config.getString(HighAvailabilityOptions.HA_CLUSTER_ID);
+		String storagePath = config.getString(HighAvailabilityOptions.HA_STORAGE_PATH) + "/" + clusterId;
+		Random rand = new Random();
+
+		try (
+			BlobServer server0 = new BlobServer(config, blobStore);
+			BlobServer server1 = new BlobServer(config, blobStore);
+			// use VoidBlobStore as the HA store to force download from each server's HA store
+			BlobCache cache0 = new BlobCache(
+				new InetSocketAddress("localhost", server0.getPort()), config,
+				new VoidBlobStore());
+			BlobCache cache1 = new BlobCache(
+				new InetSocketAddress("localhost", server1.getPort()), config,
+				new VoidBlobStore())) {
+
+			server0.start();
+			server1.start();
+
+			// Random data
+			byte[] expected = new byte[1024];
+			rand.nextBytes(expected);
+			byte[] expected2 = Arrays.copyOfRange(expected, 32, 288);
+
+			BlobKey[] keys = new BlobKey[2];
+			BlobKey nonHAKey;
+
+			// Put job-related HA data
+			JobID[] jobId = new JobID[] { new JobID(), new JobID() };
+			keys[0] = put(cache0, jobId[0], expected, true); // Request 1
+			keys[1] = put(cache0, jobId[1], expected2, true); // Request 2
+
+			// put non-HA data
+			nonHAKey = put(cache0, jobId[0], expected2, false);
+			assertNotEquals(keys[0], nonHAKey);
+			assertEquals(keys[1], nonHAKey);
+
+			// check that the storage directory exists
+			final Path blobServerPath = new Path(storagePath, "blob");
+			FileSystem fs = blobServerPath.getFileSystem();
+			assertTrue("Unknown storage dir: " + blobServerPath, fs.exists(blobServerPath));
+
+			// Verify HA requests from cache1 (connected to server1) with no immediate access to the file
+			verifyContents(cache1, jobId[0], keys[0], expected, true);
+			verifyContents(cache1, jobId[1], keys[1], expected2, true);
+
+			// Verify non-HA file is not accessible from server1
+			verifyDeleted(cache1, jobId[0], nonHAKey, true);
+		}
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/071e27f7/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheRetriesTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheRetriesTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheRetriesTest.java
index 0060ccb..9f8bfe3 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheRetriesTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheRetriesTest.java
@@ -28,13 +28,13 @@ import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
 
-import java.io.File;
+import javax.annotation.Nullable;
+
 import java.io.IOException;
-import java.io.InputStream;
 import java.net.InetSocketAddress;
-import java.net.URL;
 
-import static org.junit.Assert.*;
+import static org.apache.flink.runtime.blob.BlobServerPutTest.verifyContents;
+import static org.junit.Assert.fail;
 
 /**
  * Unit tests for the blob cache retrying the connection to the server.
@@ -54,7 +54,7 @@ public class BlobCacheRetriesTest extends TestLogger {
 		config.setString(BlobServerOptions.STORAGE_DIRECTORY,
 			temporaryFolder.newFolder().getAbsolutePath());
 
-		testBlobFetchRetries(config, new VoidBlobStore(), null);
+		testBlobFetchRetries(config, new VoidBlobStore(), null, false);
 	}
 
 	/**
@@ -67,16 +67,7 @@ public class BlobCacheRetriesTest extends TestLogger {
 		config.setString(BlobServerOptions.STORAGE_DIRECTORY,
 			temporaryFolder.newFolder().getAbsolutePath());
 
-		testBlobFetchRetries(config, new VoidBlobStore(), new JobID());
-	}
-
-	/**
-	 * A test where the connection fails twice and then the get operation succeeds
-	 * (with high availability set, job-unrelated blob).
-	 */
-	@Test
-	public void testBlobNoJobFetchRetriesHa() throws IOException {
-		testBlobFetchRetriesHa(null);
+		testBlobFetchRetries(config, new VoidBlobStore(), new JobID(), false);
 	}
 
 	/**
@@ -85,10 +76,6 @@ public class BlobCacheRetriesTest extends TestLogger {
 	 */
 	@Test
 	public void testBlobFetchRetriesHa() throws IOException {
-		testBlobFetchRetriesHa(new JobID());
-	}
-
-	private void testBlobFetchRetriesHa(final JobID jobId) throws IOException {
 		final Configuration config = new Configuration();
 		config.setString(BlobServerOptions.STORAGE_DIRECTORY,
 			temporaryFolder.newFolder().getAbsolutePath());
@@ -101,7 +88,7 @@ public class BlobCacheRetriesTest extends TestLogger {
 		try {
 			blobStoreService = BlobUtils.createBlobStoreFromConfig(config);
 
-			testBlobFetchRetries(config, blobStoreService, jobId);
+			testBlobFetchRetries(config, blobStoreService, new JobID(), true);
 		} finally {
 			if (blobStoreService != null) {
 				blobStoreService.closeAndCleanupAllData();
@@ -118,50 +105,23 @@ public class BlobCacheRetriesTest extends TestLogger {
 	 * 		set compared to this one)
 	 */
 	private static void testBlobFetchRetries(
-			final Configuration config, final BlobStore blobStore, final JobID jobId)
-			throws IOException {
-		final byte[] data = new byte[] {1, 2, 3, 4, 5, 6, 7, 8, 9, 0};
+			final Configuration config, final BlobStore blobStore, @Nullable final JobID jobId,
+			boolean highAvailability) throws IOException {
 
-		BlobServer server = null;
-		BlobCache cache = null;
-		try {
+		final byte[] data = new byte[] {1, 2, 3, 4, 5, 6, 7, 8, 9, 0};
 
-			server = new TestingFailingBlobServer(config, blobStore, 2);
+		try (
+			BlobServer server = new TestingFailingBlobServer(config, blobStore, 2);
+			BlobCache cache = new BlobCache(new InetSocketAddress("localhost", server.getPort()),
+				config, new VoidBlobStore())) {
 
-			final InetSocketAddress
-				serverAddress = new InetSocketAddress("localhost", server.getPort());
+			server.start();
 
 			// upload some blob
-			BlobClient blobClient = null;
-			BlobKey key;
-			try {
-				blobClient = new BlobClient(serverAddress, config);
-
-				key = blobClient.put(jobId, data);
-			}
-			finally {
-				if (blobClient != null) {
-					blobClient.close();
-				}
-			}
-
-			cache = new BlobCache(serverAddress, config, new VoidBlobStore());
+			final BlobKey key = BlobServerPutTest.put(server, jobId, data, highAvailability);
 
 			// trigger a download - it should fail the first two times, but retry, and succeed eventually
-			File file = jobId == null ? cache.getFile(key) : cache.getFile(jobId, key);
-			URL url = file.toURI().toURL();
-			try (InputStream is = url.openStream()) {
-				byte[] received = new byte[data.length];
-				assertEquals(data.length, is.read(received));
-				assertArrayEquals(data, received);
-			}
-		} finally {
-			if (cache != null) {
-				cache.close();
-			}
-			if (server != null) {
-				server.close();
-			}
+			verifyContents(cache, jobId, key, data, highAvailability);
 		}
 	}
 
@@ -175,7 +135,7 @@ public class BlobCacheRetriesTest extends TestLogger {
 		config.setString(BlobServerOptions.STORAGE_DIRECTORY,
 			temporaryFolder.newFolder().getAbsolutePath());
 
-		testBlobFetchWithTooManyFailures(config, new VoidBlobStore(), null);
+		testBlobFetchWithTooManyFailures(config, new VoidBlobStore(), null, false);
 	}
 
 	/**
@@ -188,28 +148,15 @@ public class BlobCacheRetriesTest extends TestLogger {
 		config.setString(BlobServerOptions.STORAGE_DIRECTORY,
 			temporaryFolder.newFolder().getAbsolutePath());
 
-		testBlobFetchWithTooManyFailures(config, new VoidBlobStore(), new JobID());
+		testBlobFetchWithTooManyFailures(config, new VoidBlobStore(), new JobID(), false);
 	}
 
 	/**
-	 * A test where the connection fails twice and then the get operation succeeds
-	 * (with high availability set, job-unrelated blob).
-	 */
-	@Test
-	public void testBlobNoJobFetchWithTooManyFailuresHa() throws IOException {
-		testBlobFetchWithTooManyFailuresHa(null);
-	}
-
-	/**
-	 * A test where the connection fails twice and then the get operation succeeds
+	 * A test where the connection fails too often and eventually fails the GET request
 	 * (with high availability set, job-related blob).
 	 */
 	@Test
 	public void testBlobForJobFetchWithTooManyFailuresHa() throws IOException {
-		testBlobFetchWithTooManyFailuresHa(new JobID());
-	}
-
-	private void testBlobFetchWithTooManyFailuresHa(final JobID jobId) throws IOException {
 		final Configuration config = new Configuration();
 		config.setString(BlobServerOptions.STORAGE_DIRECTORY,
 			temporaryFolder.newFolder().getAbsolutePath());
@@ -222,7 +169,7 @@ public class BlobCacheRetriesTest extends TestLogger {
 		try {
 			blobStoreService = BlobUtils.createBlobStoreFromConfig(config);
 
-			testBlobFetchWithTooManyFailures(config, blobStoreService, jobId);
+			testBlobFetchWithTooManyFailures(config, blobStoreService, new JobID(), true);
 		} finally {
 			if (blobStoreService != null) {
 				blobStoreService.closeAndCleanupAllData();
@@ -239,55 +186,29 @@ public class BlobCacheRetriesTest extends TestLogger {
 	 * 		set compared to this one)
 	 */
 	private static void testBlobFetchWithTooManyFailures(
-		final Configuration config, final BlobStore blobStore, final JobID jobId)
-			throws IOException {
-		final byte[] data = new byte[] { 1, 2, 3, 4, 5, 6, 7, 8, 9, 0 };
+			final Configuration config, final BlobStore blobStore, @Nullable final JobID jobId,
+			boolean highAvailability) throws IOException {
 
-		BlobServer server = null;
-		BlobCache cache = null;
-		try {
+		final byte[] data = new byte[] { 1, 2, 3, 4, 5, 6, 7, 8, 9, 0 };
 
-			server = new TestingFailingBlobServer(config, blobStore, 10);
+		try (
+			BlobServer server = new TestingFailingBlobServer(config, blobStore, 0, 10);
+			BlobCache cache = new BlobCache(new InetSocketAddress("localhost", server.getPort()),
+				config, new VoidBlobStore())) {
 
-			final InetSocketAddress
-				serverAddress = new InetSocketAddress("localhost", server.getPort());
+			server.start();
 
 			// upload some blob
-			BlobClient blobClient = null;
-			BlobKey key;
-			try {
-				blobClient = new BlobClient(serverAddress, config);
-
-				key = blobClient.put(jobId, data);
-			}
-			finally {
-				if (blobClient != null) {
-					blobClient.close();
-				}
-			}
-
-			cache = new BlobCache(serverAddress, config, new VoidBlobStore());
+			final BlobKey key = BlobServerPutTest.put(server, jobId, data, highAvailability);
 
 			// trigger a download - it should fail eventually
 			try {
-				if (jobId == null) {
-					cache.getFile(key);
-				} else {
-					cache.getFile(jobId, key);
-				}
+				verifyContents(cache, jobId, key, data, highAvailability);
 				fail("This should fail");
 			}
 			catch (IOException e) {
 				// as we expected
 			}
 		}
-		finally {
-			if (cache != null) {
-				cache.close();
-			}
-			if (server != null) {
-				server.close();
-			}
-		}
 	}
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/071e27f7/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheSuccessTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheSuccessTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheSuccessTest.java
index 1216be2..7a69804 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheSuccessTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobCacheSuccessTest.java
@@ -22,22 +22,20 @@ import org.apache.flink.api.common.JobID;
 import org.apache.flink.configuration.BlobServerOptions;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.HighAvailabilityOptions;
-import org.apache.flink.util.Preconditions;
 import org.apache.flink.util.TestLogger;
 
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
 
-import java.io.File;
+import javax.annotation.Nullable;
+
 import java.io.IOException;
 import java.net.InetSocketAddress;
-import java.util.ArrayList;
-import java.util.List;
+import java.util.Arrays;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
+import static org.apache.flink.runtime.blob.BlobServerPutTest.put;
+import static org.apache.flink.runtime.blob.BlobServerPutTest.verifyContents;
 
 /**
  * This class contains unit tests for the {@link BlobCache}.
@@ -57,7 +55,7 @@ public class BlobCacheSuccessTest extends TestLogger {
 		config.setString(BlobServerOptions.STORAGE_DIRECTORY,
 			temporaryFolder.newFolder().getAbsolutePath());
 
-		uploadFileGetTest(config, null, false, false);
+		uploadFileGetTest(config, null, false, false, false);
 	}
 
 	/**
@@ -70,76 +68,40 @@ public class BlobCacheSuccessTest extends TestLogger {
 		config.setString(BlobServerOptions.STORAGE_DIRECTORY,
 			temporaryFolder.newFolder().getAbsolutePath());
 
-		uploadFileGetTest(config, new JobID(), false, false);
-	}
-
-	/**
-	 * BlobCache is configured in HA mode and the cache can download files from
-	 * the file system directly and does not need to download BLOBs from the
-	 * BlobServer. Using job-unrelated BLOBs.
-	 */
-	@Test
-	public void testBlobNoJobCacheHa() throws IOException {
-		testBlobCacheHa(null);
+		uploadFileGetTest(config, new JobID(), false, false, false);
 	}
 
 	/**
 	 * BlobCache is configured in HA mode and the cache can download files from
 	 * the file system directly and does not need to download BLOBs from the
-	 * BlobServer. Using job-related BLOBs.
+	 * BlobServer which remains active after the BLOB upload. Using job-related BLOBs.
 	 */
 	@Test
 	public void testBlobForJobCacheHa() throws IOException {
-		testBlobCacheHa(new JobID());
-	}
-
-	private void testBlobCacheHa(final JobID jobId) throws IOException {
 		Configuration config = new Configuration();
 		config.setString(BlobServerOptions.STORAGE_DIRECTORY,
 			temporaryFolder.newFolder().getAbsolutePath());
 		config.setString(HighAvailabilityOptions.HA_MODE, "ZOOKEEPER");
 		config.setString(HighAvailabilityOptions.HA_STORAGE_PATH,
 			temporaryFolder.newFolder().getPath());
-		uploadFileGetTest(config, jobId, true, true);
-	}
 
-	/**
-	 * BlobCache is configured in HA mode and the cache can download files from
-	 * the file system directly and does not need to download BLOBs from the
-	 * BlobServer. Using job-unrelated BLOBs.
-	 */
-	@Test
-	public void testBlobNoJobCacheHa2() throws IOException {
-		testBlobCacheHa2(null);
+		uploadFileGetTest(config, new JobID(), true, true, true);
 	}
 
 	/**
 	 * BlobCache is configured in HA mode and the cache can download files from
 	 * the file system directly and does not need to download BLOBs from the
-	 * BlobServer. Using job-related BLOBs.
+	 * BlobServer which is shut down after the BLOB upload. Using job-related BLOBs.
 	 */
 	@Test
 	public void testBlobForJobCacheHa2() throws IOException {
-		testBlobCacheHa2(new JobID());
-	}
-
-	private void testBlobCacheHa2(JobID jobId) throws IOException {
 		Configuration config = new Configuration();
 		config.setString(BlobServerOptions.STORAGE_DIRECTORY,
 			temporaryFolder.newFolder().getAbsolutePath());
 		config.setString(HighAvailabilityOptions.HA_MODE, "ZOOKEEPER");
 		config.setString(HighAvailabilityOptions.HA_STORAGE_PATH,
 			temporaryFolder.newFolder().getPath());
-		uploadFileGetTest(config, jobId, false, true);
-	}
-
-	/**
-	 * BlobCache is configured in HA mode but the cache itself cannot access the
-	 * file system and thus needs to download BLOBs from the BlobServer. Using job-unrelated BLOBs.
-	 */
-	@Test
-	public void testBlobNoJobCacheHaFallback() throws IOException {
-		testBlobCacheHaFallback(null);
+		uploadFileGetTest(config, new JobID(), false, true, true);
 	}
 
 	/**
@@ -148,17 +110,14 @@ public class BlobCacheSuccessTest extends TestLogger {
 	 */
 	@Test
 	public void testBlobForJobCacheHaFallback() throws IOException {
-		testBlobCacheHaFallback(new JobID());
-	}
-
-	private void testBlobCacheHaFallback(final JobID jobId) throws IOException {
 		Configuration config = new Configuration();
 		config.setString(BlobServerOptions.STORAGE_DIRECTORY,
 			temporaryFolder.newFolder().getAbsolutePath());
 		config.setString(HighAvailabilityOptions.HA_MODE, "ZOOKEEPER");
 		config.setString(HighAvailabilityOptions.HA_STORAGE_PATH,
 			temporaryFolder.newFolder().getPath());
-		uploadFileGetTest(config, jobId, false, false);
+
+		uploadFileGetTest(config, new JobID(), false, false, true);
 	}
 
 	/**
@@ -174,101 +133,61 @@ public class BlobCacheSuccessTest extends TestLogger {
 	 * @param cacheHasAccessToFs
 	 * 		whether the cache should have access to a shared <tt>HA_STORAGE_PATH</tt> (only useful with
 	 * 		HA mode)
+	 * @param highAvailability
+	 * 		whether to use HA BLOB upload and download methods
 	 */
-	private void uploadFileGetTest(final Configuration config, JobID jobId, boolean shutdownServerAfterUpload,
-			boolean cacheHasAccessToFs) throws IOException {
-		Preconditions.checkArgument(!shutdownServerAfterUpload || cacheHasAccessToFs);
+	private void uploadFileGetTest(
+			final Configuration config, @Nullable JobID jobId, boolean shutdownServerAfterUpload,
+			boolean cacheHasAccessToFs, boolean highAvailability) throws IOException {
 
-		// First create two BLOBs and upload them to BLOB server
-		final byte[] buf = new byte[128];
-		final List<BlobKey> blobKeys = new ArrayList<BlobKey>(2);
-
-		BlobServer blobServer = null;
-		BlobCache blobCache = null;
-		BlobStoreService blobStoreService = null;
-		try {
-			final Configuration cacheConfig = new Configuration(config);
+		final Configuration cacheConfig = new Configuration(config);
+		cacheConfig.setString(BlobServerOptions.STORAGE_DIRECTORY,
+			temporaryFolder.newFolder().getAbsolutePath());
+		if (!cacheHasAccessToFs) {
+			// make sure the cache cannot access the HA store directly
 			cacheConfig.setString(BlobServerOptions.STORAGE_DIRECTORY,
 				temporaryFolder.newFolder().getAbsolutePath());
-			if (!cacheHasAccessToFs) {
-				// make sure the cache cannot access the HA store directly
-				cacheConfig.setString(BlobServerOptions.STORAGE_DIRECTORY,
-					temporaryFolder.newFolder().getAbsolutePath());
-				cacheConfig.setString(HighAvailabilityOptions.HA_STORAGE_PATH,
-					temporaryFolder.newFolder().getPath() + "/does-not-exist");
-			}
-
-			blobStoreService = BlobUtils.createBlobStoreFromConfig(cacheConfig);
-
-			// Start the BLOB server
-			blobServer = new BlobServer(config, blobStoreService);
-			final InetSocketAddress serverAddress = new InetSocketAddress(blobServer.getPort());
+			cacheConfig.setString(HighAvailabilityOptions.HA_STORAGE_PATH,
+				temporaryFolder.newFolder().getPath() + "/does-not-exist");
+		}
 
-			// Upload BLOBs
-			BlobClient blobClient = null;
-			try {
+		// First create two BLOBs and upload them to BLOB server
+		final byte[] data = new byte[128];
+		byte[] data2 = Arrays.copyOf(data, data.length);
+		data2[0] ^= 1;
 
-				blobClient = new BlobClient(serverAddress, config);
+		BlobStoreService blobStoreService = null;
 
-				blobKeys.add(blobClient.put(jobId, buf));
-				buf[0] = 1; // Make sure the BLOB key changes
-				blobKeys.add(blobClient.put(jobId, buf));
-			} finally {
-				if (blobClient != null) {
-					blobClient.close();
-				}
-			}
+		try {
+			blobStoreService = BlobUtils.createBlobStoreFromConfig(cacheConfig);
+			try (
+				BlobServer server = new BlobServer(config, blobStoreService);
+				BlobCache cache = new BlobCache(new InetSocketAddress("localhost", server.getPort()),
+					cacheConfig, blobStoreService)) {
 
-			if (shutdownServerAfterUpload) {
-				// Now, shut down the BLOB server, the BLOBs must still be accessible through the cache.
-				blobServer.close();
-				blobServer = null;
-			}
+				server.start();
 
-			blobCache = new BlobCache(serverAddress, cacheConfig, blobStoreService);
+				// Upload BLOBs
+				BlobKey key1 = put(server, jobId, data, highAvailability);
+				BlobKey key2 = put(server, jobId, data2, highAvailability);
 
-			for (BlobKey blobKey : blobKeys) {
-				if (jobId == null) {
-					blobCache.getFile(blobKey);
-				} else {
-					blobCache.getFile(jobId, blobKey);
+				if (shutdownServerAfterUpload) {
+					// Now, shut down the BLOB server, the BLOBs must still be accessible through the cache.
+					server.close();
 				}
-			}
 
-			if (blobServer != null) {
-				// Now, shut down the BLOB server, the BLOBs must still be accessible through the cache.
-				blobServer.close();
-				blobServer = null;
-			}
+				verifyContents(cache, jobId, key1, data, highAvailability);
+				verifyContents(cache, jobId, key2, data2, highAvailability);
 
-			final File[] files = new File[blobKeys.size()];
+				if (shutdownServerAfterUpload) {
+					// Now, shut down the BLOB server, the BLOBs must still be accessible through the cache.
+					server.close();
 
-			for(int i = 0; i < blobKeys.size(); i++){
-				if (jobId == null) {
-					files[i] = blobCache.getFile(blobKeys.get(i));
-				} else {
-					files[i] = blobCache.getFile(jobId, blobKeys.get(i));
+					verifyContents(cache, jobId, key1, data, highAvailability);
+					verifyContents(cache, jobId, key2, data2, highAvailability);
 				}
 			}
-
-			// Verify the result
-			assertEquals(blobKeys.size(), files.length);
-
-			for (final File file : files) {
-				assertNotNull(file);
-
-				assertTrue(file.exists());
-				assertEquals(buf.length, file.length());
-			}
 		} finally {
-			if (blobServer != null) {
-				blobServer.close();
-			}
-
-			if(blobCache != null){
-				blobCache.close();
-			}
-
 			if (blobStoreService != null) {
 				blobStoreService.closeAndCleanupAllData();
 			}