HDFS-15389. DFSAdmin should close filesystem and dfsadmin -setBalancerBandwidth should work with ViewFSOverloadScheme. Contributed by Ayush Saxena

(cherry picked from commit cc671b16f7b0b7c1ed7b41b96171653dc43cf670)
(cherry picked from commit bee2846bee4ae676bdc14585f8a3927a9dd7df37)

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/DFSAdmin.java

@@ -479,9 +479,9 @@ public class DFSAdmin extends FsShell {
   public DFSAdmin(Configuration conf) {
     super(conf);
   }
-  
+
   protected DistributedFileSystem getDFS() throws IOException {
-    return AdminHelper.getDFS(getConf());
+    return AdminHelper.checkAndGetDFS(getFS(), getConf());
   }
   
   /**
@@ -1036,14 +1036,7 @@ public class DFSAdmin extends FsShell {
       System.err.println("Bandwidth should be a non-negative integer");
       return exitCode;
     }
-
-    FileSystem fs = getFS();
-    if (!(fs instanceof DistributedFileSystem)) {
-      System.err.println("FileSystem is " + fs.getUri());
-      return exitCode;
-    }
-
-    DistributedFileSystem dfs = (DistributedFileSystem) fs;
+    DistributedFileSystem dfs = getDFS();
     try{
       dfs.setBalancerBandwidth(bandwidth);
       System.out.println("Balancer bandwidth is set to " + bandwidth);

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/tools/TestViewFileSystemOverloadSchemeWithDFSAdmin.java

@@ -263,4 +263,24 @@ public class TestViewFileSystemOverloadSchemeWithDFSAdmin {
     assertOutMsg("Disallowing snapshot on / succeeded", 1);
     assertEquals(0, ret);
   }
+
+  /**
+   * Tests setBalancerBandwidth with ViewFSOverloadScheme.
+   */
+  @Test
+  public void testSetBalancerBandwidth() throws Exception {
+    final Path hdfsTargetPath = new Path(defaultFSURI + HDFS_USER_FOLDER);
+    addMountLinks(defaultFSURI.getAuthority(),
+        new String[] {HDFS_USER_FOLDER, LOCAL_FOLDER },
+        new String[] {hdfsTargetPath.toUri().toString(),
+            localTargetDir.toURI().toString() },
+        conf);
+    final DFSAdmin dfsAdmin = new DFSAdmin(conf);
+    redirectStream();
+    int ret = ToolRunner.run(dfsAdmin,
+        new String[] {"-fs", defaultFSURI.toString(), "-setBalancerBandwidth",
+            "1000"});
+    assertOutMsg("Balancer bandwidth is set to 1000", 0);
+    assertEquals(0, ret);
+  }
 }