HDFS-14979 Allow Balancer to submit getBlocks calls to Observer Nodes when possible. Contributed by Erik Krogen.

(cherry picked from 586defe7113ed246ed0275bb3833882a3d873d70)
(cherry picked from dec765b329d3947f30273c0e7f0c4eb607ec42c9)
(cherry picked from 5cf36aa2b19aad8677e1a7553ff1f05805f772b7)

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/protocol/NamenodeProtocol.java

@@ -25,6 +25,7 @@ import org.apache.hadoop.hdfs.DFSConfigKeys;
 import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
 import org.apache.hadoop.hdfs.security.token.block.ExportedBlockKeys;
 import org.apache.hadoop.hdfs.server.namenode.CheckpointSignature;
+import org.apache.hadoop.hdfs.server.namenode.ha.ReadOnly;
 import org.apache.hadoop.io.retry.AtMostOnce;
 import org.apache.hadoop.io.retry.Idempotent;
 import org.apache.hadoop.security.KerberosInfo;
@@ -76,6 +77,7 @@ public interface NamenodeProtocol {
                                    datanode does not exist
    */
   @Idempotent
+  @ReadOnly
   public BlocksWithLocations getBlocks(DatanodeInfo datanode, long size)
   throws IOException;
 

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/balancer/TestBalancerWithHANameNodes.java

@@ -18,8 +18,14 @@
 package org.apache.hadoop.hdfs.server.balancer;
 
 import static org.junit.Assert.assertEquals;
+import static org.mockito.Matchers.anyLong;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
 import java.net.URI;
+import java.util.ArrayList;
 import java.util.Collection;
+import java.util.List;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
@@ -32,9 +38,13 @@ import org.apache.hadoop.hdfs.MiniDFSNNTopology.NNConf;
 import org.apache.hadoop.hdfs.NameNodeProxies;
 import org.apache.hadoop.hdfs.client.HdfsClientConfigKeys;
 import org.apache.hadoop.hdfs.protocol.ClientProtocol;
+import org.apache.hadoop.hdfs.protocol.DatanodeID;
 import org.apache.hadoop.hdfs.qjournal.MiniQJMHACluster;
+import org.apache.hadoop.hdfs.server.namenode.FSNamesystem;
+import org.apache.hadoop.hdfs.server.namenode.NameNodeAdapter;
 import org.apache.hadoop.hdfs.server.namenode.ha.HATestUtil;
 import org.apache.hadoop.hdfs.server.namenode.ha.ObserverReadProxyProvider;
+import org.mockito.Matchers;
 import org.junit.Test;
 
 /**
@@ -128,12 +138,24 @@ public class TestBalancerWithHANameNodes {
       cluster = qjmhaCluster.getDfsCluster();
       cluster.waitClusterUp();
       cluster.waitActive();
+      List<FSNamesystem> namesystemSpies = new ArrayList<>();
+      for (int i = 0; i < cluster.getNumNameNodes(); i++) {
+        namesystemSpies.add(
+            NameNodeAdapter.spyOnNamesystem(cluster.getNameNode(i)));
+      }
 
       DistributedFileSystem dfs = HATestUtil.configureObserverReadFs(
           cluster, conf, ObserverReadProxyProvider.class, true);
       client = dfs.getClient().getNamenode();
 
       doTest(conf);
+      for (int i = 0; i < cluster.getNumNameNodes(); i++) {
+        // First observer node is at idx 2 so it should get both getBlocks calls
+        // all other NameNodes should see 0 getBlocks calls
+        int expectedCount = (i == 2) ? 2 : 0;
+        verify(namesystemSpies.get(i), times(expectedCount))
+            .getBlocks(Matchers.<DatanodeID>any(), anyLong());
+      }
     } finally {
       if (qjmhaCluster != null) {
         qjmhaCluster.shutdown();