diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java index cbcfc553791..0383024b691 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java +++ b/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 toPortMap( + List ports) { + Map 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> 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 " + diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestSCMNodeManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestSCMNodeManager.java index 59a4938e8d7..83f99467bed 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestSCMNodeManager.java +++ b/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. */