HDDS-9846. Datanode should not persist cluster ID to global version file until loading all volumes. (apache#5757)

(cherry picked from commit 027febe)

Author: ashishkumar50

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/VersionEndpointTask.java

@@ -76,10 +76,6 @@ public EndpointStateMachine.EndPointStates call() throws Exception {
           // If end point is passive, datanode does not need to check volumes.
           String scmId = response.getValue(OzoneConsts.SCM_ID);
           String clusterId = response.getValue(OzoneConsts.CLUSTER_ID);
-          DatanodeLayoutStorage layoutStorage
-              = new DatanodeLayoutStorage(configuration);
-          layoutStorage.setClusterId(clusterId);
-          layoutStorage.persistCurrentState();
 
           Preconditions.checkNotNull(scmId,
               "Reply from SCM: scmId cannot be null");
@@ -92,6 +88,11 @@ public EndpointStateMachine.EndPointStates call() throws Exception {
           // Check HddsVolumes
           checkVolumeSet(ozoneContainer.getVolumeSet(), scmId, clusterId);
 
+          DatanodeLayoutStorage layoutStorage
+              = new DatanodeLayoutStorage(configuration);
+          layoutStorage.setClusterId(clusterId);
+          layoutStorage.persistCurrentState();
+
           // Start the container services after getting the version information
           ozoneContainer.start(clusterId);
         }

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/ozone/container/common/TestEndPoint.java

@@ -25,6 +25,7 @@
 import java.util.Map;
 import java.util.UUID;
 
+import org.apache.commons.io.FileUtils;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
@@ -79,6 +80,7 @@
 import org.junit.jupiter.api.io.TempDir;
 import org.mockito.Mockito;
 
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
@@ -285,6 +287,32 @@ public void testDnLayoutVersionFile() throws Exception {
           = new DatanodeLayoutStorage(ozoneConf,
           "na_expect_storage_initialized");
       assertEquals(scmServerImpl.getClusterId(), layout.getClusterID());
+
+      // Delete storage volume info
+      File storageDir = ozoneContainer.getVolumeSet()
+          .getVolumesList().get(0).getStorageDir();
+      FileUtils.forceDelete(storageDir);
+
+      // Format volume VERSION file with
+      // different clusterId than SCM clusterId.
+      ozoneContainer.getVolumeSet().getVolumesList()
+          .get(0).format("different_cluster_id");
+      // Update layout clusterId and persist it.
+      layout.setClusterId("different_cluster_id");
+      layout.persistCurrentState();
+
+      // As the volume level clusterId didn't match with SCM clusterId
+      // Even after the version call, the datanode layout file should
+      // not update its clusterID field.
+      rpcEndPoint.setState(EndpointStateMachine.EndPointStates.GETVERSION);
+      versionTask.call();
+      DatanodeLayoutStorage layout1
+          = new DatanodeLayoutStorage(ozoneConf,
+          "na_expect_storage_initialized");
+
+      assertEquals("different_cluster_id", layout1.getClusterID());
+      assertNotEquals(scmServerImpl.getClusterId(), layout1.getClusterID());
+      FileUtils.forceDelete(storageDir);
     }
   }