You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by st...@apache.org on 2017/08/29 18:04:31 UTC
hadoop git commit: HADOOP-14583. wasb throws an exception if you try
to create a file and there's no parent directory Contributed by Esfandiar
Manii and Thomas Marquardt.
Repository: hadoop
Updated Branches:
refs/heads/trunk 021974f4c -> 9374f3882
HADOOP-14583. wasb throws an exception if you try to create a file and there's no parent directory
Contributed by Esfandiar Manii and Thomas Marquardt.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/9374f388
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/9374f388
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/9374f388
Branch: refs/heads/trunk
Commit: 9374f3882044b552b7dbde788ce569452072c6dc
Parents: 021974f
Author: Steve Loughran <st...@apache.org>
Authored: Tue Aug 29 19:03:44 2017 +0100
Committer: Steve Loughran <st...@apache.org>
Committed: Tue Aug 29 19:03:44 2017 +0100
----------------------------------------------------------------------
.../fs/azure/AzureNativeFileSystemStore.java | 36 ++--
...estNativeAzureFileSystemConcurrencyLive.java | 165 +++++++++++++------
2 files changed, 136 insertions(+), 65 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/9374f388/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/AzureNativeFileSystemStore.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/AzureNativeFileSystemStore.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/AzureNativeFileSystemStore.java
index b0cd701..bd8ac68 100644
--- a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/AzureNativeFileSystemStore.java
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/AzureNativeFileSystemStore.java
@@ -2027,24 +2027,30 @@ public class AzureNativeFileSystemStore implements NativeFileSystemStore {
LOG.debug("Found {} as an explicit blob. Checking if it's a file or folder.", key);
- // The blob exists, so capture the metadata from the blob
- // properties.
- blob.downloadAttributes(getInstrumentedContext());
- BlobProperties properties = blob.getProperties();
+ try {
+ // The blob exists, so capture the metadata from the blob
+ // properties.
+ blob.downloadAttributes(getInstrumentedContext());
+ BlobProperties properties = blob.getProperties();
- if (retrieveFolderAttribute(blob)) {
- LOG.debug("{} is a folder blob.", key);
- return new FileMetadata(key, properties.getLastModified().getTime(),
- getPermissionStatus(blob), BlobMaterialization.Explicit);
- } else {
+ if (retrieveFolderAttribute(blob)) {
+ LOG.debug("{} is a folder blob.", key);
+ return new FileMetadata(key, properties.getLastModified().getTime(),
+ getPermissionStatus(blob), BlobMaterialization.Explicit);
+ } else {
- LOG.debug("{} is a normal blob.", key);
+ LOG.debug("{} is a normal blob.", key);
- return new FileMetadata(
- key, // Always return denormalized key with metadata.
- getDataLength(blob, properties),
- properties.getLastModified().getTime(),
- getPermissionStatus(blob));
+ return new FileMetadata(
+ key, // Always return denormalized key with metadata.
+ getDataLength(blob, properties),
+ properties.getLastModified().getTime(),
+ getPermissionStatus(blob));
+ }
+ } catch(StorageException e){
+ if (!NativeAzureFileSystemHelper.isFileNotFoundException(e)) {
+ throw e;
+ }
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/9374f388/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestNativeAzureFileSystemConcurrencyLive.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestNativeAzureFileSystemConcurrencyLive.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestNativeAzureFileSystemConcurrencyLive.java
index ec72cce..7c5899d 100644
--- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestNativeAzureFileSystemConcurrencyLive.java
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestNativeAzureFileSystemConcurrencyLive.java
@@ -19,101 +19,166 @@
package org.apache.hadoop.fs.azure;
+import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.junit.Assert;
import org.junit.Test;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
/***
* Test class to hold all Live Azure storage concurrency tests.
*/
public class TestNativeAzureFileSystemConcurrencyLive
extends AbstractWasbTestBase {
- private static final int TEST_COUNT = 102;
+ private static final int THREAD_COUNT = 102;
+ private static final int TEST_EXECUTION_TIMEOUT = 5000;
@Override
protected AzureBlobStorageTestAccount createTestAccount() throws Exception {
return AzureBlobStorageTestAccount.create();
}
/**
- * Test multi-threaded deletes in WASB. Expected behavior is one of the thread
- * should be to successfully delete the file and return true and all other
- * threads need to return false.
+ * Validate contract for FileSystem.create when overwrite is true and there
+ * are concurrent callers of FileSystem.delete. An existing file should be
+ * overwritten, even if the original destination exists but is deleted by an
+ * external agent during the create operation.
*/
- @Test
- public void testMultiThreadedDeletes() throws Exception {
+ @Test(timeout = TEST_EXECUTION_TIMEOUT)
+ public void testConcurrentCreateDeleteFile() throws Exception {
Path testFile = new Path("test.dat");
- fs.create(testFile).close();
- int threadCount = TEST_COUNT;
- DeleteHelperThread[] helperThreads = new DeleteHelperThread[threadCount];
+ List<CreateFileTask> tasks = new ArrayList<>(THREAD_COUNT);
- for (int i = 0; i < threadCount; i++) {
- helperThreads[i] = new DeleteHelperThread(fs, testFile);
+ for (int i = 0; i < THREAD_COUNT; i++) {
+ tasks.add(new CreateFileTask(fs, testFile));
}
- Thread[] threads = new Thread[threadCount];
+ ExecutorService es = null;
+
+ try {
+ es = Executors.newFixedThreadPool(THREAD_COUNT);
+
+ List<Future<Void>> futures = es.invokeAll(tasks);
+
+ for (Future<Void> future : futures) {
+ Assert.assertTrue(future.isDone());
- for (int i = 0; i < threadCount; i++) {
- threads[i] = new Thread(helperThreads[i]);
- threads[i].start();
+ // we are using Callable<V>, so if an exception
+ // occurred during the operation, it will be thrown
+ // when we call get
+ Assert.assertEquals(null, future.get());
+ }
+ } finally {
+ if (es != null) {
+ es.shutdownNow();
+ }
}
+ }
+
+ /**
+ * Validate contract for FileSystem.delete when invoked concurrently.
+ * One of the threads should successfully delete the file and return true;
+ * all other threads should return false.
+ */
+ @Test(timeout = TEST_EXECUTION_TIMEOUT)
+ public void testConcurrentDeleteFile() throws Exception {
+ Path testFile = new Path("test.dat");
+ fs.create(testFile).close();
- for (int i = 0; i < threadCount; i++) {
- threads[i].join();
+ List<DeleteFileTask> tasks = new ArrayList<>(THREAD_COUNT);
+
+ for (int i = 0; i < THREAD_COUNT; i++) {
+ tasks.add(new DeleteFileTask(fs, testFile));
}
- boolean deleteSuccess = false;
+ ExecutorService es = null;
+ try {
+ es = Executors.newFixedThreadPool(THREAD_COUNT);
+
+ List<Future<Boolean>> futures = es.invokeAll(tasks);
- for (int i = 0; i < threadCount; i++) {
+ int successCount = 0;
+ for (Future<Boolean> future : futures) {
+ Assert.assertTrue(future.isDone());
- Assert.assertFalse("child thread has exception : " + helperThreads[i].getException(),
- helperThreads[i].getExceptionEncounteredFlag());
+ // we are using Callable<V>, so if an exception
+ // occurred during the operation, it will be thrown
+ // when we call get
+ Boolean success = future.get();
+ if (success) {
+ successCount++;
+ }
+ }
- if (deleteSuccess) {
- Assert.assertFalse("More than one thread delete() retuhelperThreads[i].getDeleteSuccess()",
- helperThreads[i].getExceptionEncounteredFlag());
- } else {
- deleteSuccess = helperThreads[i].getDeleteSuccess();
+ Assert.assertEquals(
+ "Exactly one delete operation should return true.",
+ 1,
+ successCount);
+ } finally {
+ if (es != null) {
+ es.shutdownNow();
}
}
-
- Assert.assertTrue("No successfull delete found", deleteSuccess);
}
}
-class DeleteHelperThread implements Runnable {
+abstract class FileSystemTask<V> implements Callable<V> {
+ private final FileSystem fileSystem;
+ private final Path path;
+
+ protected FileSystem getFileSystem() {
+ return this.fileSystem;
+ }
- private FileSystem fs;
- private Path p;
- private boolean deleteSuccess;
- private boolean exceptionEncountered;
- private Exception ex;
+ protected Path getFilePath() {
+ return this.path;
+ }
- public DeleteHelperThread(FileSystem fs, Path p) {
- this.fs = fs;
- this.p = p;
+ FileSystemTask(FileSystem fs, Path p) {
+ this.fileSystem = fs;
+ this.path = p;
}
- public void run() {
- try {
- deleteSuccess = fs.delete(p, false);
- } catch (Exception ioEx) {
- exceptionEncountered = true;
- this.ex = ioEx;
- }
+ public abstract V call() throws Exception;
+}
+
+class DeleteFileTask extends FileSystemTask<Boolean> {
+
+ DeleteFileTask(FileSystem fs, Path p) {
+ super(fs, p);
}
- public boolean getDeleteSuccess() {
- return deleteSuccess;
+ @Override
+ public Boolean call() throws Exception {
+ return this.getFileSystem().delete(this.getFilePath(), false);
}
+}
- public boolean getExceptionEncounteredFlag() {
- return exceptionEncountered;
+class CreateFileTask extends FileSystemTask<Void> {
+ CreateFileTask(FileSystem fs, Path p) {
+ super(fs, p);
}
- public Exception getException() {
- return ex;
+ public Void call() throws Exception {
+ FileSystem fs = getFileSystem();
+ Path p = getFilePath();
+
+ // Create an empty file and close the stream.
+ FSDataOutputStream stream = fs.create(p, true);
+ stream.close();
+
+ // Delete the file. We don't care if delete returns true or false.
+ // We just want to ensure the file does not exist.
+ this.getFileSystem().delete(this.getFilePath(), false);
+
+ return null;
}
}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org