HBASE-7439 HFileLink should not use the configuration from the Filesystem (Matteo Bertozzi)

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

hbase-server/src/main/java/org/apache/hadoop/hbase/io/HFileLink.java

@@ -124,19 +124,6 @@ public class HFileLink extends FileLink {
     return m.groupCount() > 2 && m.group(2) != null && m.group(3) != null;
   }
 
-  /**
-   * The returned path can be the "original" file path like: /hbase/table/region/cf/hfile
-   * or a path to the archived file like: /hbase/archive/table/region/cf/hfile
-   *
-   * @param fs {@link FileSystem} on which to check the HFileLink
-   * @param path HFileLink path
-   * @return Referenced path (original path or archived path)
-   * @throws IOException on unexpected error.
-   */
-  public static Path getReferencedPath(FileSystem fs, final Path path) throws IOException {
-    return getReferencedPath(fs.getConf(), fs, path);
-  }
-
   /**
    * The returned path can be the "original" file path like: /hbase/table/region/cf/hfile
    * or a path to the archived file like: /hbase/.archive/table/region/cf/hfile

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

@@ -156,6 +156,8 @@ public class StoreFile {
   // If this storefile is a link to another, this is the link instance.
   private HFileLink link;
 
+  private Configuration conf;
+
   // Block cache configuration and reference.
   private final CacheConfig cacheConf;
 
@@ -257,6 +259,7 @@ public class StoreFile {
       throws IOException {
     this.fs = fs;
     this.path = p;
+    this.conf = conf;
     this.cacheConf = cacheConf;
     this.dataBlockEncoder =
         dataBlockEncoder == null ? NoOpDataBlockEncoder.INSTANCE
@@ -513,28 +516,6 @@ public class StoreFile {
     return FSUtils.computeHDFSBlocksDistribution(fs, status, start, length);
   }
 
-  /**
-   * helper function to compute HDFS blocks distribution of a given file.
-   * For reference file, it is an estimate
-   * @param fs  The FileSystem
-   * @param p  The path of the file
-   * @return HDFS blocks distribution
-   */
-  static public HDFSBlocksDistribution computeHDFSBlockDistribution(
-    FileSystem fs, Path p) throws IOException {
-    if (isReference(p)) {
-      Reference reference = Reference.read(fs, p);
-      Path referencePath = getReferredToFile(p);
-      return computeRefFileHDFSBlockDistribution(fs, reference, referencePath);
-    } else {
-      if (HFileLink.isHFileLink(p)) p = HFileLink.getReferencedPath(fs, p);
-      FileStatus status = fs.getFileStatus(p);
-      long length = status.getLen();
-      return FSUtils.computeHDFSBlocksDistribution(fs, status, 0, length);
-    }
-  }
-
-
   /**
    * compute HDFS block distribution, for reference file, it is an estimate
    */
@@ -590,7 +571,7 @@ public class StoreFile {
         this.reference = Reference.read(fs, this.path);
         this.referencePath = getReferredToLink(this.path);
         LOG.debug("Reference file "+ path + " referred to " + referencePath + "!");
-        link = new HFileLink(fs.getConf(), referencePath);
+        link = new HFileLink(conf, referencePath);
         this.reader = new HalfStoreFileReader(this.fs, this.referencePath, link,
             this.cacheConf, this.reference,
             dataBlockEncoder.getEncodingInCache());

hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreFile.java

@@ -240,10 +240,13 @@ public class TestStoreFile extends HBaseTestCase {
    */
   public void testReferenceToHFileLink() throws IOException {
     final String columnFamily = "f";
+
+    Path rootDir = FSUtils.getRootDir(conf);
+
     String tablename = "_original-evil-name"; // adding legal table name chars to verify regex handles it.
     HRegionInfo hri = new HRegionInfo(Bytes.toBytes(tablename));
     // store dir = <root>/<tablename>/<rgn>/<cf>
-    Path storedir = new Path(new Path(FSUtils.getRootDir(conf),
+    Path storedir = new Path(new Path(rootDir,
       new Path(hri.getTableNameAsString(), hri.getEncodedName())), columnFamily);
 
     // Make a store file and write data to it. <root>/<tablename>/<rgn>/<cf>/<file>
@@ -257,7 +260,7 @@ public class TestStoreFile extends HBaseTestCase {
 
     // create link to store file. <root>/clone/region/<cf>/<hfile>-<region>-<table>
     String target = "clone";
-    Path dstPath = new Path(FSUtils.getRootDir(conf), new Path(new Path(target, "region"), columnFamily));
+    Path dstPath = new Path(rootDir, new Path(new Path(target, "region"), columnFamily));
     HFileLink.create(conf, this.fs, dstPath, hri, storeFilePath.getName());
     Path linkFilePath = new Path(dstPath,
                   HFileLink.createHFileLinkName(hri, storeFilePath.getName()));
@@ -265,9 +268,9 @@ public class TestStoreFile extends HBaseTestCase {
     // create splits of the link.
     // <root>/clone/splitA/<cf>/<reftohfilelink>,
     // <root>/clone/splitB/<cf>/<reftohfilelink>
-    Path splitDirA = new Path(new Path(FSUtils.getRootDir(conf),
+    Path splitDirA = new Path(new Path(rootDir,
         new Path(target, "splitA")), columnFamily);
-    Path splitDirB = new Path(new Path(FSUtils.getRootDir(conf),
+    Path splitDirB = new Path(new Path(rootDir,
         new Path(target, "splitB")), columnFamily);
     StoreFile f = new StoreFile(fs, linkFilePath, conf, cacheConf, BloomType.NONE,
         NoOpDataBlockEncoder.INSTANCE);
@@ -276,7 +279,7 @@ public class TestStoreFile extends HBaseTestCase {
     Path pathB = StoreFile.split(fs, splitDirB, f, splitRow, false); // bottom
 
     // OK test the thing
-    FSUtils.logFileSystemState(fs, FSUtils.getRootDir(conf), LOG);
+    FSUtils.logFileSystemState(fs, rootDir, LOG);
 
     // There is a case where a file with the hfilelink pattern is actually a daughter
     // reference to a hfile link.  This code in StoreFile that handles this case.