HDDS-15887. SCM should update DatanodeDetails ports on DN re-registration or heartbeat when only ports change

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java

@@ -475,6 +475,7 @@ public RegisteredCommand register(
               datanodeDetails, oldNode.getVersion(), datanodeDetails.getVersion());
           nodeStateManager.updateNode(datanodeDetails, layoutInfo);
         }
+        updateDatanodePortsIfChanged(datanodeDetails, layoutInfo);
       } catch (NodeNotFoundException e) {
         LOG.error("Cannot find datanode {} from nodeStateManager",
                 datanodeDetails);
@@ -547,6 +548,33 @@ private boolean isVersionChange(String oldVersion, String newVersion) {
     return versionChanged;
   }
 
+  private boolean isPortChange(DatanodeDetails oldNode, DatanodeDetails newNode) {
+    return !toPortMap(oldNode.getPorts()).equals(toPortMap(newNode.getPorts()));
+  }
+
+  private Map<DatanodeDetails.Port.Name, Integer> toPortMap(
+      List<DatanodeDetails.Port> ports) {
+    Map<DatanodeDetails.Port.Name, Integer> portMap = new HashMap<>();
+    for (DatanodeDetails.Port port : ports) {
+      portMap.put(port.getName(), port.getValue());
+    }
+    return portMap;
+  }
+
+  private void updateDatanodePortsIfChanged(DatanodeDetails datanodeDetails,
+      LayoutVersionProto layoutInfo) throws NodeNotFoundException {
+    DatanodeInfo oldNode = nodeStateManager.getNode(datanodeDetails);
+    if (!isPortChange(oldNode, datanodeDetails)) {
+      return;
+    }
+    LOG.info("Update the ports for registered datanode {}, " +
+            "oldPorts = {}, newPorts = {}.",
+        datanodeDetails, oldNode.getPorts(), datanodeDetails.getPorts());
+    nodeStateManager.updateNode(datanodeDetails, layoutInfo);
+    DatanodeDetails dn = nodeStateManager.getNode(datanodeDetails);
+    scmNodeEventPublisher.fireEvent(SCMEvents.NODE_ADDRESS_UPDATE, dn);
+  }
+
   /**
    * Send heartbeat to indicate the datanode is alive and doing well.
    *
@@ -562,6 +590,9 @@ public List<SCMCommand<?>> processHeartbeat(DatanodeDetails datanodeDetails,
       nodeStateManager.updateLastHeartbeatTime(datanodeDetails);
       metrics.incNumHBProcessed();
       updateDatanodeOpState(datanodeDetails);
+      DatanodeInfo oldNode = nodeStateManager.getNode(datanodeDetails);
+      updateDatanodePortsIfChanged(datanodeDetails,
+          oldNode.getLastKnownLayoutVersion());
     } catch (NodeNotFoundException e) {
       metrics.incNumHBProcessingFailed();
       LOG.error("SCM trying to process heartbeat from an " +

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestSCMNodeManager.java

@@ -38,6 +38,7 @@
 import static org.apache.hadoop.hdds.scm.events.SCMEvents.DATANODE_COMMAND;
 import static org.apache.hadoop.hdds.scm.events.SCMEvents.DATANODE_COMMAND_COUNT_UPDATED;
 import static org.apache.hadoop.hdds.scm.events.SCMEvents.NEW_NODE;
+import static org.apache.hadoop.hdds.scm.events.SCMEvents.NODE_ADDRESS_UPDATE;
 import static org.apache.hadoop.ozone.container.upgrade.UpgradeUtils.toLayoutVersionProto;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -2102,6 +2103,89 @@ public void testScmRegisterNodeWithUpdatedIpAndHostname()
     }
   }
 
+  /**
+   * Test node register with updated ports only.
+   */
+  @Test
+  public void testScmRegisterNodeWithUpdatedPorts()
+      throws IOException, NodeNotFoundException, AuthenticationException {
+    OzoneConfiguration conf = getConf();
+    SCMStorageConfig scmStorageConfig = mock(SCMStorageConfig.class);
+    when(scmStorageConfig.getClusterID()).thenReturn("xyz111");
+    EventPublisher eventPublisher = mock(EventPublisher.class);
+    HDDSLayoutVersionManager lvm =
+        new HDDSLayoutVersionManager(scmStorageConfig.getLayoutVersion());
+    SCMContext nodeManagerContext = SCMContext.emptyContext();
+    try (SCMNodeManager nodeManager = new SCMNodeManager(conf,
+        scmStorageConfig, eventPublisher, new NetworkTopologyImpl(conf),
+        nodeManagerContext, lvm)) {
+      DatanodeID nodeId = DatanodeID.randomID();
+      DatanodeDetails node = createDatanodeWithPorts(nodeId, 9855, false);
+      nodeManager.register(node, null, null, CORRECT_LAYOUT_PROTO);
+
+      DatanodeDetails storedNode = nodeManager.getNodeStateManager().getNode(node);
+      assertTrue(storedNode.getPorts().stream()
+          .noneMatch(p -> p.getName() == DatanodeDetails.Port.Name.RATIS_DATASTREAM));
+
+      DatanodeDetails updatedNode = createDatanodeWithPorts(nodeId, 9855, true);
+      nodeManager.register(updatedNode, null, null, CORRECT_LAYOUT_PROTO);
+
+      storedNode = nodeManager.getNodeStateManager().getNode(node);
+      assertTrue(storedNode.hasPort(DatanodeDetails.Port.Name.RATIS_DATASTREAM));
+      assertEquals(9855,
+          storedNode.getPort(DatanodeDetails.Port.Name.RATIS_DATASTREAM).getValue());
+      verify(eventPublisher, times(1)).fireEvent(NODE_ADDRESS_UPDATE, storedNode);
+    }
+  }
+
+  /**
+   * Test heartbeat updates ports when only ports change.
+   */
+  @Test
+  public void testScmHeartbeatWithUpdatedPorts()
+      throws IOException, NodeNotFoundException, AuthenticationException {
+    OzoneConfiguration conf = getConf();
+    SCMStorageConfig scmStorageConfig = mock(SCMStorageConfig.class);
+    when(scmStorageConfig.getClusterID()).thenReturn("xyz111");
+    EventPublisher eventPublisher = mock(EventPublisher.class);
+    HDDSLayoutVersionManager lvm =
+        new HDDSLayoutVersionManager(scmStorageConfig.getLayoutVersion());
+    SCMContext nodeManagerContext = SCMContext.emptyContext();
+    try (SCMNodeManager nodeManager = new SCMNodeManager(conf,
+        scmStorageConfig, eventPublisher, new NetworkTopologyImpl(conf),
+        nodeManagerContext, lvm)) {
+      DatanodeID nodeId = DatanodeID.randomID();
+      DatanodeDetails node = createDatanodeWithPorts(nodeId, 9855, false);
+      nodeManager.register(node, null, null, CORRECT_LAYOUT_PROTO);
+      nodeManager.processLayoutVersionReport(node, CORRECT_LAYOUT_PROTO);
+
+      DatanodeDetails updatedNode = createDatanodeWithPorts(nodeId, 9855, true);
+      nodeManager.processHeartbeat(updatedNode, null);
+
+      DatanodeDetails storedNode = nodeManager.getNodeStateManager().getNode(node);
+      assertTrue(storedNode.hasPort(DatanodeDetails.Port.Name.RATIS_DATASTREAM));
+      assertEquals(9855,
+          storedNode.getPort(DatanodeDetails.Port.Name.RATIS_DATASTREAM).getValue());
+      verify(eventPublisher, times(1)).fireEvent(NODE_ADDRESS_UPDATE, storedNode);
+    }
+  }
+
+  private DatanodeDetails createDatanodeWithPorts(DatanodeID nodeId, int ratisPort,
+      boolean includeDatastreamPort) {
+    DatanodeDetails.Builder builder = DatanodeDetails.newBuilder()
+        .setID(nodeId)
+        .setHostName("host1")
+        .setIpAddress("1.2.3.4");
+    for (DatanodeDetails.Port.Name name : DatanodeDetails.Port.Name.V0_PORTS) {
+      builder.addPort(DatanodeDetails.newPort(name, ratisPort));
+    }
+    if (includeDatastreamPort) {
+      builder.addPort(DatanodeDetails.newPort(
+          DatanodeDetails.Port.Name.RATIS_DATASTREAM, ratisPort));
+    }
+    return builder.build();
+  }
+
   /**
    * Test that pipelineLimit() uses the default value when the config is not set.
    */