HBASE-7982 TestReplicationQueueFailover* runs for a minute, spews 3/4million lines complaining 'Filesystem closed', has an NPE, and still passes?

git-svn-id: https://svn.apache.org/repos/asf/hbase/trunk@1453712 13f79535-47bb-0310-9956-ffa450edef68

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java

@@ -1159,25 +1159,29 @@ public class HRegionServer implements ClientProtocol,
   }
 
   private void closeWAL(final boolean delete) {
-    try {
     if (this.hlogForMeta != null) {
       // All hlogs (meta and non-meta) are in the same directory. Don't call
       // closeAndDelete here since that would delete all hlogs not just the
       // meta ones. We will just 'close' the hlog for meta here, and leave
       // the directory cleanup to the follow-on closeAndDelete call.
+      try {
         this.hlogForMeta.close();
+      } catch (Throwable e) {
+        LOG.error("Metalog close and delete failed", RemoteExceptionHandler.checkThrowable(e));
+      }
     }
     if (this.hlog != null) {
+      try {
         if (delete) {
           hlog.closeAndDelete();
         } else {
           hlog.close();
         }
-      }
       } catch (Throwable e) {
         LOG.error("Close and delete failed", RemoteExceptionHandler.checkThrowable(e));
       }
     }
+  }
 
   private void closeAllScanners() {
     // Close any outstanding scanners. Means they'll get an UnknownScanner

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/HLogSplitter.java

@@ -511,7 +511,12 @@ public class HLogSplitter {
     List<Path> corruptedLogs = new ArrayList<Path>();
     FileSystem fs;
     fs = rootdir.getFileSystem(conf);
-    Path logPath = new Path(logfile);
+    Path logPath = null;
+    if (FSUtils.isStartingWithPath(rootdir, logfile)) {
+      logPath = new Path(logfile);
+    } else {
+      logPath = new Path(rootdir, logfile);
+    }
     if (ZKSplitLog.isCorrupted(rootdir, logPath.getName(), fs)) {
       corruptedLogs.add(logPath);
     } else {

hbase-server/src/main/java/org/apache/hadoop/hbase/util/FSUtils.java

@@ -86,6 +86,61 @@ public abstract class FSUtils {
     super();
   }
 
+  /**
+   * Compare of path component. Does not consider schema; i.e. if schemas different but <code>path
+   * <code> starts with <code>rootPath<code>, then the function returns true
+   * @param rootPath
+   * @param path 
+   * @return True if <code>path</code> starts with <code>rootPath</code>
+   */
+  public static boolean isStartingWithPath(final Path rootPath, final String path) {
+    String uriRootPath = rootPath.toUri().getPath();
+    String tailUriPath = (new Path(path)).toUri().getPath();
+    return tailUriPath.startsWith(uriRootPath);
+  }
+
+  /**
+   * Compare path component of the Path URI; e.g. if hdfs://a/b/c and /a/b/c, it will compare the
+   * '/a/b/c' part. Does not consider schema; i.e. if schemas different but path or subpath matches,
+   * the two will equate.
+   * @param pathToSearch Path we will be trying to match.
+   * @param pathTail
+   * @return True if <code>pathTail</code> is tail on the path of <code>pathToSearch</code>
+   */
+  public static boolean isMatchingTail(final Path pathToSearch, String pathTail) {
+    return isMatchingTail(pathToSearch, new Path(pathTail));
+  }
+
+  /**
+   * Compare path component of the Path URI; e.g. if hdfs://a/b/c and /a/b/c, it will compare the
+   * '/a/b/c' part. If you passed in 'hdfs://a/b/c and b/c, it would return true.  Does not consider
+   * schema; i.e. if schemas different but path or subpath matches, the two will equate.
+   * @param pathToSearch Path we will be trying to match.
+   * @param pathTail
+   * @return True if <code>pathTail</code> is tail on the path of <code>pathToSearch</code>
+   */
+  public static boolean isMatchingTail(final Path pathToSearch, final Path pathTail) {
+    if (pathToSearch.depth() != pathTail.depth()) return false;
+    Path tailPath = pathTail;
+    String tailName;
+    Path toSearch = pathToSearch;
+    String toSearchName;
+    boolean result = false;
+    do {
+      tailName = tailPath.getName();
+      if (tailName == null || tailName.length() <= 0) {
+        result = true;
+        break;
+      }
+      toSearchName = toSearch.getName();
+      if (toSearchName == null || toSearchName.length() <= 0) break;
+      // Move up a parent on each path for next go around.  Path doesn't let us go off the end.
+      tailPath = tailPath.getParent();
+      toSearch = toSearch.getParent();
+    } while(tailName.equals(toSearchName));
+    return result;
+  }
+
   public static FSUtils getInstance(FileSystem fs, Configuration conf) {
     String scheme = fs.getUri().getScheme();
     if (scheme == null) {

hbase-server/src/test/java/org/apache/hadoop/hbase/util/TestFSUtils.java

@@ -34,12 +34,12 @@ import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.permission.FsPermission;
-import org.apache.hadoop.hbase.exceptions.DeserializationException;
 import org.apache.hadoop.hbase.HBaseConfiguration;
 import org.apache.hadoop.hbase.HBaseTestingUtility;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.HDFSBlocksDistribution;
 import org.apache.hadoop.hbase.MediumTests;
+import org.apache.hadoop.hbase.exceptions.DeserializationException;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
@@ -49,6 +49,33 @@ import org.junit.experimental.categories.Category;
  */
 @Category(MediumTests.class)
 public class TestFSUtils {
+  /**
+   * Test path compare and prefix checking.
+   * @throws IOException 
+   */
+  @Test
+  public void testMatchingTail() throws IOException {
+    HBaseTestingUtility htu = new HBaseTestingUtility();
+    final FileSystem fs = htu.getTestFileSystem();
+    Path rootdir = htu.getDataTestDir();
+    assertTrue(rootdir.depth() > 1);
+    Path partPath = new Path("a", "b");
+    Path fullPath = new Path(rootdir, partPath);
+    Path fullyQualifiedPath = fs.makeQualified(fullPath);
+    assertFalse(FSUtils.isMatchingTail(fullPath, partPath));
+    assertFalse(FSUtils.isMatchingTail(fullPath, partPath.toString()));
+    assertTrue(FSUtils.isStartingWithPath(rootdir, fullPath.toString()));
+    assertTrue(FSUtils.isStartingWithPath(fullyQualifiedPath, fullPath.toString()));
+    assertFalse(FSUtils.isStartingWithPath(rootdir, partPath.toString()));
+    assertFalse(FSUtils.isMatchingTail(fullyQualifiedPath, partPath));
+    assertTrue(FSUtils.isMatchingTail(fullyQualifiedPath, fullPath));
+    assertTrue(FSUtils.isMatchingTail(fullyQualifiedPath, fullPath.toString()));
+    assertTrue(FSUtils.isMatchingTail(fullyQualifiedPath, fs.makeQualified(fullPath)));
+    assertTrue(FSUtils.isStartingWithPath(rootdir, fullyQualifiedPath.toString()));
+    assertFalse(FSUtils.isMatchingTail(fullPath, new Path("x")));
+    assertFalse(FSUtils.isMatchingTail(new Path("x"), fullPath));
+  }
+
   @Test
   public void testVersion() throws DeserializationException, IOException {
     HBaseTestingUtility htu = new HBaseTestingUtility();