Delete the baseTrashPath's subDir leads to don't modify baseTrashPath

Signed-off-by: sunlisheng <sunlisheng@xiaomi.com>

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/CommonConfigurationKeysPublic.java

@@ -171,6 +171,13 @@ public class CommonConfigurationKeysPublic {
   public static final String  FS_TRASH_INTERVAL_KEY = "fs.trash.interval";
   /** Default value for FS_TRASH_INTERVAL_KEY */
   public static final long    FS_TRASH_INTERVAL_DEFAULT = 0;
+
+  public static final String MOVE_TO_TRASH_FOR_TEST_KEY = "move.to.trash.for.tests";
+  /**
+   * Default value for MOVE_TO_TRASH_FOR_TEST_KEY
+   */
+  public static final boolean MOVE_TO_TRASH_FOR_TEST_DEFAULT = false;
+
   /**
    * @see
    * <a href="{@docRoot}/../hadoop-project-dist/hadoop-common/core-default.xml">

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/TrashPolicy.java

@@ -36,6 +36,7 @@ public abstract class TrashPolicy extends Configured {
   protected Path trash; // path to trash directory
   protected long deletionInterval; // deletion interval for Emptier
 
+  volatile boolean flag = false; // Control thread execution order
   /**
    * Used to setup the trash policy. Must be implemented by all TrashPolicy
    * implementations.

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/TrashPolicyDefault.java

@@ -21,6 +21,8 @@ import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_TRASH_CHECKP
 import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_TRASH_CHECKPOINT_INTERVAL_KEY;
 import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_TRASH_INTERVAL_DEFAULT;
 import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_TRASH_INTERVAL_KEY;
+import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.MOVE_TO_TRASH_FOR_TEST_DEFAULT;
+import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.MOVE_TO_TRASH_FOR_TEST_KEY;
 
 import java.io.FileNotFoundException;
 import java.io.IOException;
@@ -70,6 +72,8 @@ public class TrashPolicyDefault extends TrashPolicy {
 
   private long emptierInterval;
 
+  private volatile boolean moveToTrashForTests = false;
+
   public TrashPolicyDefault() { }
 
   private TrashPolicyDefault(FileSystem fs, Configuration conf)
@@ -107,6 +111,8 @@ public class TrashPolicyDefault extends TrashPolicy {
           + "Changing to default value 0", deletionInterval);
       this.deletionInterval = 0;
     }
+
+    this.moveToTrashForTests = conf.getBoolean(MOVE_TO_TRASH_FOR_TEST_KEY, MOVE_TO_TRASH_FOR_TEST_DEFAULT);
   }
 
   private Path makeTrashRelativePath(Path basePath, Path rmFilePath) {
@@ -157,14 +163,36 @@ public class TrashPolicyDefault extends TrashPolicy {
       } catch (FileAlreadyExistsException e) {
         // find the path which is not a directory, and modify baseTrashPath
         // & trashPath, then mkdirs
+        if (moveToTrashForTests) {
+          flag = true;
+          while (flag)
+            ;
+
+        }
         Path existsFilePath = baseTrashPath;
         while (!fs.exists(existsFilePath)) {
           existsFilePath = existsFilePath.getParent();
         }
-        baseTrashPath = new Path(baseTrashPath.toString().replace(
+        // race condition: don't modify baseTrashPath when existsFilePath is deleted.
-            existsFilePath.toString(), existsFilePath.toString() + Time.now())
+        // existsFilePath is /user/test/.Trash/Current/user/test/a/b
-        );
+        // another thread delete /user/test/.Trash/Current/user/test/a in this corner case before the code as follow:
-        trashPath = new Path(baseTrashPath, trashPath.getName());
+        // baseTrashPath = new Path(baseTrashPath.toString().replace(
+        //                  existsFilePath.toString(), existsFilePath.toString() + Time.now()));
+        // trashPath = new Path(baseTrashPath, trashPath.getName());
+        // baseTrashPath is /user/test/.Trash/Current/user/test+timestamp
+        // trashPath is /user/test/.Trash/Current/user/test+timestamp/a/b. It is not expected result.
+        try {
+          FileStatus fileStatus = fs.getFileStatus(existsFilePath);
+          if (fileStatus.isFile()) {
+            baseTrashPath = new Path(baseTrashPath.toString().replace(
+                    existsFilePath.toString(), existsFilePath.toString() + Time.now())
+            );
+            trashPath = new Path(baseTrashPath, trashPath.getName());
+          }
+        } catch (FileNotFoundException e1) {
+          LOG.warn("The existFilePath was not found " + existsFilePath);
+        }
+
         // retry, ignore current failure
         --i;
         continue;

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/TestTrash.java

@@ -53,6 +53,7 @@ import org.apache.hadoop.util.Time;
  * This class tests commands from Trash.
  */
 public class TestTrash {
+  public static final Logger LOGGER = LoggerFactory.getLogger(TestTrash.class);
 
   private final static File BASE_PATH = new File(GenericTestUtils.getTempPath(
       "testTrash"));
@@ -627,6 +628,45 @@ public class TestTrash {
           e.getLocalizedMessage());
     }
     assertTrue(val2 == 0);
+
+    // Third rm a file which parent path is the same as above.
+    // Make baseTrashPath, at the same time delete existTrashPath
+    conf = new Configuration();
+    conf.setLong(FS_TRASH_INTERVAL_KEY, 10);
+    conf.setBoolean(MOVE_TO_TRASH_FOR_TEST_KEY, true);
+    FileSystem fileSystem = FileSystem.get(conf);
+    Path existingFile = new Path("existingFile");
+    Path subFile = new Path("existingFile", "subFile");
+    fileSystem.delete(existingFile, true);
+    fileSystem.create(existingFile);
+
+    Trash trash = new Trash(fileSystem, conf);
+    // Make sure trash root is clean
+    Path trashRoot = trash.getCurrentTrashDir(existingFile);
+    fileSystem.delete(trashRoot, true);
+    // Move to trash should be succeed
+    assertTrue("The existing file to trash failed",
+        trash.moveToTrash(existingFile));
+    // Verify the existing file is removed
+    assertFalse("The existing file still exists on file system",
+        fileSystem.exists(existingFile));
+    fileSystem.create(subFile);
+
+    new Thread(new Runnable() {
+      @Override
+      public void run() {
+        try {
+          while (!trash.getTrashPolicy().flag)
+            ;
+          fileSystem.delete(trash.getCurrentTrashDir(existingFile), true);
+          trash.getTrashPolicy().flag = false;
+        } catch (IOException e) {
+          LOGGER.warn("Delete filed failed", e);
+        }
+      }
+    }).start();
+    assertTrue("Move a subFile to trash failed", trash.moveToTrash(subFile));
+    assertTrue(fileSystem.exists(trash.getCurrentTrashDir(existingFile)));
   }
 
   @Test