From 48bef285a85e311b52335c8687c6e9a5f8afdf70 Mon Sep 17 00:00:00 2001
From: YulongZ <sky.zyl@hotmail.com>
Date: Wed, 16 Feb 2022 00:44:17 +0800
Subject: [PATCH] HDFS-16440. RBF: Support router get HAServiceStatus with
 Lifeline RPC address (#3971)

---
 .../router/NamenodeHeartbeatService.java      | 10 +++-
 .../router/TestRouterNamenodeHeartbeat.java   | 56 +++++++++++++++++--
 2 files changed, 61 insertions(+), 5 deletions(-)

diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/NamenodeHeartbeatService.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/NamenodeHeartbeatService.java
index 8858341c69..e1c8a6846f 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/NamenodeHeartbeatService.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/NamenodeHeartbeatService.java
@@ -48,6 +48,8 @@ import org.codehaus.jettison.json.JSONObject;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.hadoop.classification.VisibleForTesting;
+
 /**
  * The {@link Router} periodically checks the state of a Namenode (usually on
  * the same server) and reports their high availability (HA) state and
@@ -346,7 +348,8 @@ public class NamenodeHeartbeatService extends PeriodicService {
           // Determine if NN is active
           // TODO: dynamic timeout
           if (localTargetHAProtocol == null) {
-            localTargetHAProtocol = localTarget.getProxy(conf, 30*1000);
+            localTargetHAProtocol = localTarget.getHealthMonitorProxy(conf, 30*1000);
+            LOG.debug("Get HA status with address {}", lifelineAddress);
           }
           HAServiceStatus status = localTargetHAProtocol.getServiceStatus();
           report.setHAServiceState(status.getState());
@@ -373,6 +376,11 @@ public class NamenodeHeartbeatService extends PeriodicService {
     return report;
   }
 
+  @VisibleForTesting
+  NNHAServiceTarget getLocalTarget(){
+    return this.localTarget;
+  }
+
   /**
    * Get the description of the Namenode to monitor.
    * @return Description of the Namenode to monitor.
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterNamenodeHeartbeat.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterNamenodeHeartbeat.java
index 9bf149c649..94f2baeaed 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterNamenodeHeartbeat.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterNamenodeHeartbeat.java
@@ -211,6 +211,50 @@ public class TestRouterNamenodeHeartbeat {
     assertEquals(NAMENODES[1], standby.getNamenodeId());
   }
 
+  @Test
+  public void testNamenodeHeartbeatServiceHAServiceProtocolProxy(){
+    testNamenodeHeartbeatServiceHAServiceProtocol(
+        "test-ns", "nn", 1000, -1, -1, 1003,
+        "host01.test:1000", "host02.test:1000");
+    testNamenodeHeartbeatServiceHAServiceProtocol(
+        "test-ns", "nn", 1000, 1001, -1, 1003,
+        "host01.test:1001", "host02.test:1001");
+    testNamenodeHeartbeatServiceHAServiceProtocol(
+        "test-ns", "nn", 1000, -1, 1002, 1003,
+        "host01.test:1002", "host02.test:1002");
+    testNamenodeHeartbeatServiceHAServiceProtocol(
+        "test-ns", "nn", 1000, 1001, 1002, 1003,
+        "host01.test:1002", "host02.test:1002");
+  }
+
+  private void testNamenodeHeartbeatServiceHAServiceProtocol(
+      String nsId, String nnId,
+      int rpcPort, int servicePort,
+      int lifelinePort, int webAddressPort,
+      String expected0, String expected1) {
+    Configuration conf = generateNamenodeConfiguration(nsId, nnId,
+        rpcPort, servicePort, lifelinePort, webAddressPort);
+
+    Router testRouter = new Router();
+    testRouter.setConf(conf);
+
+    Collection<NamenodeHeartbeatService> heartbeatServices =
+        testRouter.createNamenodeHeartbeatServices();
+
+    assertEquals(2, heartbeatServices.size());
+
+    Iterator<NamenodeHeartbeatService> iterator = heartbeatServices.iterator();
+    NamenodeHeartbeatService service0 = iterator.next();
+    service0.init(conf);
+    assertNotNull(service0.getLocalTarget());
+    assertEquals(expected0, service0.getLocalTarget().getHealthMonitorAddress().toString());
+
+    NamenodeHeartbeatService service1 = iterator.next();
+    service1.init(conf);
+    assertNotNull(service1.getLocalTarget());
+    assertEquals(expected1, service1.getLocalTarget().getHealthMonitorAddress().toString());
+  }
+
   @Test
   public void testNamenodeHeartbeatServiceNNResolution() {
     String nsId = "test-ns";
@@ -261,10 +305,14 @@ public class TestRouterNamenodeHeartbeat {
 
     conf.set(DFS_NAMENODE_RPC_ADDRESS_KEY + "." + suffix,
         MockDomainNameResolver.DOMAIN + ":" + rpcPort);
-    conf.set(DFS_NAMENODE_SERVICE_RPC_ADDRESS_KEY + "." + suffix,
-        MockDomainNameResolver.DOMAIN + ":" + servicePort);
-    conf.set(DFS_NAMENODE_LIFELINE_RPC_ADDRESS_KEY + "." + suffix,
-        MockDomainNameResolver.DOMAIN + ":" + lifelinePort);
+    if (servicePort >= 0){
+      conf.set(DFS_NAMENODE_SERVICE_RPC_ADDRESS_KEY + "." + suffix,
+          MockDomainNameResolver.DOMAIN + ":" + servicePort);
+    }
+    if (lifelinePort >= 0){
+      conf.set(DFS_NAMENODE_LIFELINE_RPC_ADDRESS_KEY + "." + suffix,
+          MockDomainNameResolver.DOMAIN + ":" + lifelinePort);
+    }
     conf.set(DFS_NAMENODE_HTTP_ADDRESS_KEY + "." + suffix,
         MockDomainNameResolver.DOMAIN + ":" + webAddressPort);