From 36d634d353c2193d87d60f6525647360d8a27379 Mon Sep 17 00:00:00 2001
From: Mikhail Antonov <antonov@apache.org>
Date: Mon, 2 May 2016 13:23:51 -0700
Subject: [PATCH] HBASE-15281 Allow the FileSystem inside HFileSystem to be
 wrapped (Rajesh Nishtala)

---
 .../apache/hadoop/hbase/fs/HFileSystem.java   | 27 +++++++++++++++++--
 1 file changed, 25 insertions(+), 2 deletions(-)

diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/fs/HFileSystem.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/fs/HFileSystem.java
index fb5836055d9..521bc572b5e 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/fs/HFileSystem.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/fs/HFileSystem.java
@@ -101,11 +101,13 @@ public class HFileSystem extends FilterFileSystem {
     if (useHBaseChecksum && !(fs instanceof LocalFileSystem)) {
       conf = new Configuration(conf);
       conf.setBoolean("dfs.client.read.shortcircuit.skip.checksum", true);
-      this.noChecksumFs = newInstanceFileSystem(conf);
+      this.noChecksumFs = maybeWrapFileSystem(newInstanceFileSystem(conf), conf);
       this.noChecksumFs.setVerifyChecksum(false);
     } else {
-      this.noChecksumFs = fs;
+      this.noChecksumFs = maybeWrapFileSystem(fs, conf);
     }
+
+    this.fs = maybeWrapFileSystem(this.fs, conf);
   }
 
   /**
@@ -191,6 +193,27 @@ public class HFileSystem extends FilterFileSystem {
     return fs;
   }
 
+  /**
+   * Returns an instance of Filesystem wrapped into the class specified in
+   * hbase.fs.wrapper property, if one is set in the configuration, returns
+   * unmodified FS instance passed in as an argument otherwise.
+   * @param base Filesystem instance to wrap
+   * @param conf Configuration
+   * @return wrapped instance of FS, or the same instance if no wrapping configured.
+   */
+  private FileSystem maybeWrapFileSystem(FileSystem base, Configuration conf) {
+    try {
+      Class<?> clazz = conf.getClass("hbase.fs.wrapper", null);
+      if (clazz != null) {
+        return (FileSystem) clazz.getConstructor(FileSystem.class, Configuration.class)
+          .newInstance(base, conf);
+      }
+    } catch (Exception e) {
+      LOG.error("Failed to wrap filesystem: " + e);
+    }
+    return base;
+  }
+
   public static boolean addLocationsOrderInterceptor(Configuration conf) throws IOException {
     return addLocationsOrderInterceptor(conf, new ReorderWALBlocks());
   }