@Override
  public void run() {
    try {
      Path rootHDFSDirPath = new Path(rootHDFSDir);
      FileSystem fs = FileSystem.get(rootHDFSDirPath.toUri(), conf);
      FileStatus[] userHDFSDirList = fs.listStatus(rootHDFSDirPath);

      List<Path> scratchDirToRemove = new ArrayList<Path>();
      for (FileStatus userHDFSDir : userHDFSDirList) {
        FileStatus[] scratchDirList = fs.listStatus(userHDFSDir.getPath());
        for (FileStatus scratchDir : scratchDirList) {
          Path lockFilePath = new Path(scratchDir.getPath(), SessionState.LOCK_FILE_NAME);
          if (!fs.exists(lockFilePath)) {
            String message = "Skipping " + scratchDir.getPath() + " since it does not contain " +
                SessionState.LOCK_FILE_NAME;
            if (verbose) {
              consoleMessage(message);
            }
            continue;
          }
          boolean removable = false;
          boolean inuse = false;
          try {
            IOUtils.closeStream(fs.append(lockFilePath));
            removable = true;
          } catch (RemoteException eAppend) {
            // RemoteException with AlreadyBeingCreatedException will be thrown
            // if the file is currently held by a writer
            if(AlreadyBeingCreatedException.class.getName().equals(eAppend.getClassName())){
              inuse = true;
            } else {
              consoleMessage("Unexpected error:" + eAppend.getMessage());
            }
          } catch (UnsupportedOperationException eUnsupported) {
            // In Hadoop-3, append method is not supported.
            // This is an alternative check to make sure whether a file is in use or not.
            // Trying to open the file. If it is in use, it will throw IOException.
            try {
              IOUtils.closeStream(fs.create(lockFilePath, false));
            } catch (RemoteException eCreate) {
              if (AlreadyBeingCreatedException.class.getName().equals(eCreate.getClassName())){
                // If the file is held by a writer, will throw AlreadyBeingCreatedException
                inuse = true;
              }  else {
                consoleMessage("Unexpected error:" + eCreate.getMessage());
              }
            } catch (FileAlreadyExistsException eCreateNormal) {
              // Otherwise, throw FileAlreadyExistsException, which means the file owner is dead
              removable = true;
            }
          }
          if (inuse) {
            // Cannot open the lock file for writing, must be held by a live process
            String message = scratchDir.getPath() + " is being used by live process";
            if (verbose) {
              consoleMessage(message);
            }
          }
          if (removable) {
            scratchDirToRemove.add(scratchDir.getPath());
          }
        }
      }

      if (scratchDirToRemove.size()==0) {
        consoleMessage("Cannot find any scratch directory to clear");
        return;
      }
      consoleMessage("Removing " + scratchDirToRemove.size() + " scratch directories");
      String localTmpDir = HiveConf.getVar(conf, HiveConf.ConfVars.LOCAL_SCRATCH_DIR);
      for (Path scratchDir : scratchDirToRemove) {
        if (dryRun) {
          System.out.println(scratchDir);
        } else {
          boolean succ = fs.delete(scratchDir, true);
          if (!succ) {
            consoleMessage("Cannot remove " + scratchDir);
          } else {
            String message = scratchDir + " removed";
            if (verbose) {
              consoleMessage(message);
            }
          }
          // cleaning up on local file system as well
          removeLocalTmpFiles(scratchDir.getName(), localTmpDir);
        }
      }
    } catch (IOException e) {
      consoleMessage("Unexpected exception " + e.getMessage());
    }
  }
A tool to remove dangling scratch directory. A scratch directory could be left behind
in some cases, such as when vm restarts and leave no chance for Hive to run shutdown hook.
The tool will test a scratch directory is use, if not, remove it.
We rely on HDFS write lock for to detect if a scratch directory is in use:
1. A HDFS client open HDFS file ($scratchdir/inuse.lck) for write and only close
   it at the time the session is closed
2. cleardanglingscratchDir can try to open $scratchdir/inuse.lck for write. If the
   corresponding HiveCli/HiveServer2 is still running, we will get exception.
   Otherwise, we know the session is dead
3. If the HiveCli/HiveServer2 dies without closing the HDFS file, NN will reclaim the
   lease after 10 min, ie, the HDFS file hold by the dead HiveCli/HiveServer2 is writable
   again after 10 min. Once it become writable, cleardanglingscratchDir will be able to
   remove it
4. Additional functionality; once it is decided which session scratch dirs are residual,
   while removing them from hdfs, we will remove them from local tmp location as well.
   Please see {@link ClearDanglingScratchDir#removeLocalTmpFiles(String, String)}.
ql/src/java/org/apache/hadoop/hive/ql/session/ClearDanglingScratchDir.java