HDDS-9884. Pass DatanodeVersion to the client

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocol/DatanodeDetails.java

@@ -372,6 +372,12 @@ public static DatanodeDetails.Builder newBuilder(
       builder.setPersistedOpStateExpiry(
           datanodeDetailsProto.getPersistedOpStateExpiry());
     }
+    if (datanodeDetailsProto.hasInitialVersion()) {
+      builder.setInitialVersion(datanodeDetailsProto.getInitialVersion());
+    }
+    if (datanodeDetailsProto.hasCurrentVersion()) {
+      builder.setCurrentVersion(datanodeDetailsProto.getCurrentVersion());
+    }
     return builder;
   }
 
@@ -476,6 +482,9 @@ public HddsProtos.DatanodeDetailsProto.Builder toProtoBuilder(
       }
     }
 
+    builder.setInitialVersion(initialVersion);
+    builder.setCurrentVersion(currentVersion);
+
     return builder;
   }
 

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/HddsDatanodeService.java

@@ -125,6 +125,9 @@ public class HddsDatanodeService extends GenericCli implements ServicePlugin {
   private HddsDatanodeClientProtocolServer clientProtocolServer;
   private OzoneAdmins admins;
   private ReconfigurationHandler reconfigurationHandler;
+  // Default datanode initial and current version to be used when datanode.id file doesn't exist
+  // Note: Currently this is only used in tests.
+  private int defaultInitialVersion = -1, defaultCurrentVersion = -1;
 
   //Constructor for DataNode PluginService
   public HddsDatanodeService() { }
@@ -232,7 +235,6 @@ public void start() {
       datanodeDetails.setRevision(
           HddsVersionInfo.HDDS_VERSION_INFO.getRevision());
       datanodeDetails.setBuildDate(HddsVersionInfo.HDDS_VERSION_INFO.getDate());
-      datanodeDetails.setCurrentVersion(DatanodeVersion.CURRENT_VERSION);
       TracingUtil.initTracing(
           "HddsDatanodeService." + datanodeDetails.getUuidString()
               .substring(0, 8), conf);
@@ -422,14 +424,15 @@ private DatanodeDetails initializeDatanodeDetails()
     Preconditions.checkNotNull(idFilePath);
     File idFile = new File(idFilePath);
     if (idFile.exists()) {
-      return ContainerUtils.readDatanodeDetailsFrom(idFile);
+      DatanodeDetails details = ContainerUtils.readDatanodeDetailsFrom(idFile);
+      details.setCurrentVersion(DatanodeVersion.CURRENT_VERSION);
+      return details;
     } else {
       // There is no datanode.id file, this might be the first time datanode
       // is started.
-      DatanodeDetails details = DatanodeDetails.newBuilder()
-          .setUuid(UUID.randomUUID()).build();
-      details.setInitialVersion(DatanodeVersion.CURRENT_VERSION);
-      details.setCurrentVersion(DatanodeVersion.CURRENT_VERSION);
+      DatanodeDetails details = DatanodeDetails.newBuilder().setUuid(UUID.randomUUID()).build();
+      details.setInitialVersion(defaultInitialVersion >= 0 ? defaultInitialVersion : DatanodeVersion.CURRENT_VERSION);
+      details.setCurrentVersion(defaultCurrentVersion >= 0 ? defaultCurrentVersion : DatanodeVersion.CURRENT_VERSION);
       return details;
     }
   }
@@ -667,4 +670,20 @@ private String reconfigDeletingServiceWorkers(String value) {
         .setPoolSize(Integer.parseInt(value));
     return value;
   }
+
+  public int getDefaultInitialVersion() {
+    return defaultInitialVersion;
+  }
+
+  public void setDefaultInitialVersion(int defaultInitialVersion) {
+    this.defaultInitialVersion = defaultInitialVersion;
+  }
+
+  public int getDefaultCurrentVersion() {
+    return defaultCurrentVersion;
+  }
+
+  public void setDefaultCurrentVersion(int defaultCurrentVersion) {
+    this.defaultCurrentVersion = defaultCurrentVersion;
+  }
 }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/DatanodeIdYaml.java

@@ -56,7 +56,7 @@ private DatanodeIdYaml() {
   }
 
   /**
-   * Creates a yaml file using DatnodeDetails. This method expects the path
+   * Creates a yaml file using DatanodeDetails. This method expects the path
    * validation to be performed by the caller.
    *
    * @param datanodeDetails {@link DatanodeDetails}

hadoop-hdds/interface-client/src/main/proto/hdds.proto

@@ -45,6 +45,8 @@ message DatanodeDetailsProto {
     optional string networkLocation = 7; // Network topology location
     optional NodeOperationalState persistedOpState = 8; // The Operational state persisted in the datanode.id file
     optional int64 persistedOpStateExpiry = 9; // The seconds after the epoch when the OpState should expire
+    optional int32 initialVersion = 10;  // Initial datanode version. TODO: This is not useful for clients and can be ignored.
+    optional int32 currentVersion = 11;  // Current datanode wire version
     // TODO(runzhiwang): when uuid is gone, specify 1 as the index of uuid128 and mark as required
     optional UUID uuid128 = 100; // UUID with 128 bits assigned to the Datanode.
 }

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneCluster.java

@@ -18,6 +18,7 @@
 package org.apache.hadoop.ozone;
 
 import java.io.IOException;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Optional;
 import java.util.OptionalInt;
@@ -338,11 +339,13 @@ abstract class Builder {
     protected Optional<StorageUnit> streamBufferSizeUnit = Optional.empty();
     protected boolean includeRecon = false;
 
-
     protected Optional<Integer> omLayoutVersion = Optional.empty();
     protected Optional<Integer> scmLayoutVersion = Optional.empty();
     protected Optional<Integer> dnLayoutVersion = Optional.empty();
 
+    protected Optional<int[]> dnInitialVersion = Optional.empty();
+    protected Optional<int[]> dnCurrentVersion = Optional.empty();
+
     // Use relative smaller number of handlers for testing
     protected int numOfOmHandlers = 20;
     protected int numOfScmHandlers = 20;
@@ -455,6 +458,60 @@ public Builder setNumDatanodes(int val) {
       return this;
     }
 
+    /**
+     * Set the (same) initialVersion for all datanodes.
+     * Caution: Call setNumDatanodes() before this if you need to.
+     *
+     * @param val initialVersion value to be set for all datanodes.
+     *
+     * @return MiniOzoneCluster.Builder
+     */
+    public Builder setDatanodeInitialVersion(int val) {
+      int[] arr = new int[numDataVolumes];
+      Arrays.fill(arr, val);
+      dnInitialVersion = Optional.of(arr);
+      return this;
+    }
+
+    /**
+     * Set the (same) currentVersion for all datanodes.
+     * Caution: Call setNumDatanodes() before this if you need to.
+     *
+     * @param val currentVersion value to be set for all datanodes.
+     *
+     * @return MiniOzoneCluster.Builder
+     */
+    public Builder setDatanodeCurrentVersion(int val) {
+      int[] arr = new int[numDataVolumes];
+      Arrays.fill(arr, val);
+      dnCurrentVersion = Optional.of(arr);
+      return this;
+    }
+
+    /**
+     * Set initialVersion for each datanode separately.
+     *
+     * @param varArray an array of initialVersion for each datanode
+     *
+     * @return MiniOzoneCluster.Builder
+     */
+    public Builder setDatanodeInitialVersion(int[] varArray) {
+      dnInitialVersion = Optional.of(varArray);
+      return this;
+    }
+
+    /**
+     * Set currentVersion for each datanode separately.
+     *
+     * @param varArray an array of currentVersion for each datanode
+     *
+     * @return MiniOzoneCluster.Builder
+     */
+    public Builder setDatanodeCurrentVersion(int[] varArray) {
+      dnCurrentVersion = Optional.of(varArray);
+      return this;
+    }
+
     /**
      * Sets the number of data volumes per datanode.
      *

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneClusterImpl.java

@@ -138,23 +138,32 @@ public class MiniOzoneClusterImpl implements MiniOzoneCluster {
   private final Set<AutoCloseable> clients = ConcurrentHashMap.newKeySet();
   private SecretKeyClient secretKeyClient;
 
+  // TODO: Get rid of Optional, just use null
+  private Optional<int[]> dnInitialVersion = Optional.empty();
+  private Optional<int[]> dnCurrentVersion = Optional.empty();
+
   /**
    * Creates a new MiniOzoneCluster with Recon.
    *
    * @throws IOException if there is an I/O error
    */
+  @SuppressWarnings("checkstyle:ParameterNumber")
   MiniOzoneClusterImpl(OzoneConfiguration conf,
                        SCMConfigurator scmConfigurator,
                        OzoneManager ozoneManager,
                        StorageContainerManager scm,
                        List<HddsDatanodeService> hddsDatanodes,
-                       ReconServer reconServer) {
+                       ReconServer reconServer,
+                       Optional<int[]> dnInitialVersion,
+                       Optional<int[]> dnCurrentVersion) {
     this.conf = conf;
     this.ozoneManager = ozoneManager;
     this.scm = scm;
     this.hddsDatanodes = hddsDatanodes;
     this.reconServer = reconServer;
     this.scmConfigurator = scmConfigurator;
+    this.dnInitialVersion = dnInitialVersion;
+    this.dnCurrentVersion = dnCurrentVersion;
   }
 
   /**
@@ -327,6 +336,14 @@ public int getHddsDatanodeIndex(DatanodeDetails dn) throws IOException {
         "Not able to find datanode with datanode Id " + dn.getUuid());
   }
 
+  public int getDatanodeInitialVersion(int dnIdx) {
+    return dnInitialVersion.map(v -> v[dnIdx]).orElse(-1);
+  }
+
+  public int getDatanodeCurrentVersion(int dnIdx) {
+    return dnCurrentVersion.map(v -> v[dnIdx]).orElse(-1);
+  }
+
   @Override
   public OzoneClient newClient() throws IOException {
     OzoneClient client = createClient();
@@ -416,7 +433,7 @@ public void restartHddsDatanode(int i, boolean waitForDatanode)
     HddsDatanodeService service = new HddsDatanodeService(args);
     service.setConfiguration(config);
     hddsDatanodes.add(i, service);
-    startHddsDatanode(service);
+    startHddsDatanode(service, i);
     if (waitForDatanode) {
       // wait for the node to be identified as a healthy node again.
       waitForClusterToBeReady();
@@ -477,13 +494,19 @@ public void startScm() throws IOException {
     scm.start();
   }
 
-  public void startHddsDatanode(HddsDatanodeService datanode) {
+  public void startHddsDatanode(HddsDatanodeService datanode, int dnIdx) {
     try {
       datanode.setCertificateClient(getCAClient());
     } catch (IOException e) {
       LOG.error("Exception while setting certificate client to DataNode.", e);
     }
     datanode.setSecretKeyClient(secretKeyClient);
+    if (dnInitialVersion.isPresent()) {
+      datanode.setDefaultInitialVersion(getDatanodeInitialVersion(dnIdx));
+    }
+    if (dnCurrentVersion.isPresent()) {
+      datanode.setDefaultCurrentVersion(getDatanodeCurrentVersion(dnIdx));
+    }
     datanode.start();
   }
 
@@ -492,7 +515,9 @@ public void startHddsDatanode(HddsDatanodeService datanode) {
    */
   @Override
   public void startHddsDatanodes() {
-    hddsDatanodes.forEach(this::startHddsDatanode);
+    for (int i = 0; i < hddsDatanodes.size(); i++) {
+      startHddsDatanode(hddsDatanodes.get(i), i);
+    }
   }
 
   @Override
@@ -617,7 +642,8 @@ public MiniOzoneCluster build() throws IOException {
 
         MiniOzoneClusterImpl cluster = new MiniOzoneClusterImpl(conf,
             scmConfigurator, om, scm,
-            hddsDatanodes, reconServer);
+            hddsDatanodes, reconServer,
+            dnInitialVersion, dnCurrentVersion);
 
         cluster.setCAClient(certClient);
         cluster.setSecretKeyClient(secretKeyClient);
@@ -843,10 +869,19 @@ protected List<HddsDatanodeService> createHddsDatanodes(
       String[] args = new String[] {};
       conf.setStrings(ScmConfigKeys.OZONE_SCM_NAMES, scmAddress);
       List<HddsDatanodeService> hddsDatanodes = new ArrayList<>();
+
+      // Sanity check
+      if (dnInitialVersion.isPresent() && dnInitialVersion.get().length != numOfDatanodes) {
+        throw new IllegalArgumentException("Number of initial versions should be equal to number of datanodes");
+      }
+      if (dnCurrentVersion.isPresent() && dnCurrentVersion.get().length != numOfDatanodes) {
+        throw new IllegalArgumentException("Number of current versions should be equal to number of datanodes");
+      }
+
       for (int i = 0; i < numOfDatanodes; i++) {
         OzoneConfiguration dnConf = new OzoneConfiguration(conf);
         configureDatanodePorts(dnConf);
-        String datanodeBaseDir = path + "/datanode-" + Integer.toString(i);
+        String datanodeBaseDir = path + "/datanode-" + i;
         Path metaDir = Paths.get(datanodeBaseDir, "meta");
         List<String> dataDirs = new ArrayList<>();
         List<String> reservedSpaceList = new ArrayList<>();

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestBlockDataStreamOutput.java

@@ -20,13 +20,15 @@
 import org.apache.hadoop.conf.StorageUnit;
 import org.apache.hadoop.hdds.client.ReplicationType;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
 import org.apache.hadoop.hdds.scm.OzoneClientConfig;
 import org.apache.hadoop.hdds.scm.XceiverClientManager;
 import org.apache.hadoop.hdds.scm.XceiverClientMetrics;
 import org.apache.hadoop.hdds.scm.storage.BlockDataStreamOutput;
 import org.apache.hadoop.hdds.scm.storage.ByteBufferStreamOutput;
 import org.apache.hadoop.hdds.utils.IOUtils;
+import org.apache.hadoop.ozone.HddsDatanodeService;
 import org.apache.hadoop.ozone.MiniOzoneCluster;
 import org.apache.hadoop.ozone.OzoneConfigKeys;
 import org.apache.hadoop.ozone.client.ObjectStore;
@@ -44,6 +46,7 @@
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 
@@ -69,6 +72,8 @@ public class TestBlockDataStreamOutput {
   private static String volumeName;
   private static String bucketName;
   private static String keyString;
+  private static final int[] DN_INITIAL_VERSION = new int[] {0, 0, 0, 0, 1};
+  private static final int[] DN_CURRENT_VERSION = new int[] {0, 0, 0, 1, 1};
 
   /**
    * Create a MiniDFSCluster for testing.
@@ -94,6 +99,8 @@ public static void init() throws Exception {
 
     cluster = MiniOzoneCluster.newBuilder(conf)
         .setNumDatanodes(5)
+        .setDatanodeInitialVersion(DN_INITIAL_VERSION)
+        .setDatanodeCurrentVersion(DN_CURRENT_VERSION)
         .setTotalPipelineNumLimit(3)
         .setBlockSize(blockSize)
         .setChunkSize(chunkSize)
@@ -267,4 +274,33 @@ public void testTotalAckDataLength() throws Exception {
     assertEquals(dataLength, stream.getTotalAckDataLength());
   }
 
+  @Test
+  public void testDatanodeVersion() throws Exception {
+
+    // First verify DNs has their respective versions set correctly
+    List<HddsDatanodeService> dns = cluster.getHddsDatanodes();
+    for (int i = 0; i < dns.size(); i++) {
+      HddsDatanodeService dn = dns.get(i);
+      assertEquals(DN_INITIAL_VERSION[i], dn.getDefaultInitialVersion());
+      assertEquals(DN_CURRENT_VERSION[i], dn.getDefaultCurrentVersion());
+
+      DatanodeDetails details = dn.getDatanodeDetails();
+      assertEquals(dn.getDefaultInitialVersion(), details.getInitialVersion());
+      assertEquals(dn.getDefaultCurrentVersion(), details.getCurrentVersion());
+    }
+
+    String keyName = getKeyName();
+    OzoneDataStreamOutput key = createKey(keyName, ReplicationType.RATIS, 0);
+    KeyDataStreamOutput keyDataStreamOutput = (KeyDataStreamOutput) key.getByteBufStreamOutput();
+    BlockDataStreamOutputEntry stream = keyDataStreamOutput.getStreamEntries().get(0);
+    // Now check 3 DNs in a random pipeline have the correct DN versions
+    List<DatanodeDetails> streamDnDetails = stream.getPipeline().getNodes();
+    for (DatanodeDetails details : streamDnDetails) {
+      int dnIdx = cluster.getHddsDatanodeIndex(details);
+
+      assertEquals(DN_INITIAL_VERSION[dnIdx], details.getInitialVersion());
+      assertEquals(DN_CURRENT_VERSION[dnIdx], details.getCurrentVersion());
+    }
+  }
+
 }