HDDS-14753. Do not prematurely serialize client version

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

@@ -513,14 +513,14 @@ public static DatanodeDetails getFromProtoBuf(
    */
   @JsonIgnore
   public HddsProtos.DatanodeDetailsProto getProtoBufMessage() {
-    return toProto(ClientVersion.CURRENT.serialize());
+    return toProto(ClientVersion.CURRENT);
   }
 
-  public HddsProtos.DatanodeDetailsProto toProto(int clientVersion) {
+  public HddsProtos.DatanodeDetailsProto toProto(ClientVersion clientVersion) {
     return toProtoBuilder(clientVersion, Collections.emptySet()).build();
   }
 
-  public HddsProtos.DatanodeDetailsProto toProto(int clientVersion, Set<Port.Name> filterPorts) {
+  public HddsProtos.DatanodeDetailsProto toProto(ClientVersion clientVersion, Set<Port.Name> filterPorts) {
     return toProtoBuilder(clientVersion, filterPorts).build();
   }
 
@@ -533,7 +533,7 @@ public HddsProtos.DatanodeDetailsProto toProto(int clientVersion, Set<Port.Name>
    * @return A {@link HddsProtos.DatanodeDetailsProto.Builder} Object.
    */
   public HddsProtos.DatanodeDetailsProto.Builder toProtoBuilder(
-      int clientVersion, Set<Port.Name> filterPorts) {
+      ClientVersion clientVersion, Set<Port.Name> filterPorts) {
 
     final HddsProtos.DatanodeIDProto idProto = id.toProto();
     final HddsProtos.DatanodeDetailsProto.Builder builder =
@@ -1172,7 +1172,7 @@ public void setRevision(String rev) {
 
   @Override
   public HddsProtos.NetworkNode toProtobuf(
-      int clientVersion) {
+      ClientVersion clientVersion) {
     return HddsProtos.NetworkNode.newBuilder()
         .setDatanodeDetails(toProtoBuilder(clientVersion, Collections.emptySet()).build())
         .build();

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/common/helpers/ContainerWithPipeline.java

@@ -24,6 +24,7 @@
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 import org.apache.hadoop.hdds.scm.container.ContainerInfo;
 import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
+import org.apache.hadoop.ozone.ClientVersion;
 
 /**
  * Class wraps ozone container info.
@@ -53,7 +54,7 @@ public static ContainerWithPipeline fromProtobuf(HddsProtos.ContainerWithPipelin
         Pipeline.getFromProtobuf(allocatedContainer.getPipeline()));
   }
 
-  public HddsProtos.ContainerWithPipeline getProtobuf(int clientVersion) {
+  public HddsProtos.ContainerWithPipeline getProtobuf(ClientVersion clientVersion) {
     return HddsProtos.ContainerWithPipeline.newBuilder()
         .setContainerInfo(getContainerInfo().getProtobuf())
         .setPipeline(getPipeline().getProtobufMessage(clientVersion, Name.IO_PORTS))

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/net/InnerNode.java

@@ -20,6 +20,7 @@
 import java.util.Collection;
 import java.util.List;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.ozone.ClientVersion;
 
 /**
  * The interface defines an inner node in a network topology.
@@ -92,7 +93,7 @@ Node getLeaf(int leafIndex, List<String> excludedScopes,
       Collection<Node> excludedNodes, int ancestorGen);
 
   @Override
-  HddsProtos.NetworkNode toProtobuf(int clientVersion);
+  HddsProtos.NetworkNode toProtobuf(ClientVersion clientVersion);
 
   @Override
   boolean equals(Object o);

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/net/Node.java

@@ -18,6 +18,7 @@
 package org.apache.hadoop.hdds.scm.net;
 
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.ozone.ClientVersion;
 
 /**
  * The interface defines a node in a network topology.
@@ -130,7 +131,7 @@ public interface Node {
   boolean isDescendant(String nodePath);
 
   default HddsProtos.NetworkNode toProtobuf(
-      int clientVersion) {
+      ClientVersion clientVersion) {
     return null;
   }
 }

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/pipeline/Pipeline.java

@@ -67,7 +67,7 @@ public final class Pipeline {
   private static final Codec<Pipeline> CODEC = new DelegatedCodec<>(
       Proto2Codec.get(HddsProtos.Pipeline.getDefaultInstance()),
       Pipeline::getFromProtobufSetCreationTimestamp,
-      p -> p.getProtobufMessage(ClientVersion.CURRENT.serialize()),
+      p -> p.getProtobufMessage(ClientVersion.CURRENT),
       Pipeline.class,
       DelegatedCodec.CopyType.UNSUPPORTED);
 
@@ -363,11 +363,12 @@ public ReplicationConfig getReplicationConfig() {
     return replicationConfig;
   }
 
-  public HddsProtos.Pipeline getProtobufMessage(int clientVersion) {
+  public HddsProtos.Pipeline getProtobufMessage(ClientVersion clientVersion) {
     return getProtobufMessage(clientVersion, Collections.emptySet());
   }
 
-  public HddsProtos.Pipeline getProtobufMessage(int clientVersion, Set<DatanodeDetails.Port.Name> filterPorts) {
+  public HddsProtos.Pipeline getProtobufMessage(ClientVersion clientVersion,
+      Set<DatanodeDetails.Port.Name> filterPorts) {
     List<HddsProtos.DatanodeDetailsProto> members = new ArrayList<>();
     List<Integer> memberReplicaIndexes = new ArrayList<>();
 

hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/protocol/TestDatanodeDetails.java

@@ -32,6 +32,7 @@
 import org.apache.hadoop.hdds.protocol.DatanodeDetails.Port;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails.Port.Name;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.ozone.ClientVersion;
 import org.junit.jupiter.api.Test;
 
 /**
@@ -44,11 +45,11 @@ void protoIncludesNewPortsOnlyForV1() {
     DatanodeDetails subject = MockDatanodeDetails.randomDatanodeDetails();
 
     HddsProtos.DatanodeDetailsProto proto =
-        subject.toProto(DEFAULT_VERSION.serialize());
+        subject.toProto(DEFAULT_VERSION);
     assertPorts(proto, V0_PORTS);
 
     HddsProtos.DatanodeDetailsProto protoV1 =
-        subject.toProto(VERSION_HANDLES_UNKNOWN_DN_PORTS.serialize());
+        subject.toProto(VERSION_HANDLES_UNKNOWN_DN_PORTS);
     assertPorts(protoV1, ALL_PORTS);
   }
 
@@ -58,11 +59,11 @@ void testRequiredPortsProto() {
     Set<Port.Name> requiredPorts = Stream.of(Port.Name.STANDALONE, Port.Name.RATIS)
         .collect(Collectors.toSet());
     HddsProtos.DatanodeDetailsProto proto =
-        subject.toProto(subject.getCurrentVersion(), requiredPorts);
+        subject.toProto(ClientVersion.deserialize(subject.getCurrentVersion()), requiredPorts);
     assertPorts(proto, ImmutableSet.copyOf(requiredPorts));
 
     HddsProtos.DatanodeDetailsProto ioPortProto =
-        subject.toProto(subject.getCurrentVersion(), Name.IO_PORTS);
+        subject.toProto(ClientVersion.deserialize(subject.getCurrentVersion()), Name.IO_PORTS);
     assertPorts(ioPortProto, ImmutableSet.copyOf(Name.IO_PORTS));
   }
 
@@ -74,15 +75,15 @@ public void testNewBuilderCurrentVersion() {
     Set<Port.Name> requiredPorts = Stream.of(Port.Name.STANDALONE, Port.Name.RATIS)
         .collect(Collectors.toSet());
     HddsProtos.DatanodeDetailsProto.Builder protoBuilder =
-        dn.toProtoBuilder(DEFAULT_VERSION.serialize(), requiredPorts);
+        dn.toProtoBuilder(DEFAULT_VERSION, requiredPorts);
     protoBuilder.clearCurrentVersion();
     DatanodeDetails dn2 = DatanodeDetails.newBuilder(protoBuilder.build()).build();
     assertEquals(HDDSVersion.SEPARATE_RATIS_PORTS_AVAILABLE,
         HDDSVersion.deserialize(dn2.getCurrentVersion()));
 
     // test that if the current version is set, it is used
     protoBuilder =
-        dn.toProtoBuilder(DEFAULT_VERSION.serialize(), requiredPorts);
+        dn.toProtoBuilder(DEFAULT_VERSION, requiredPorts);
     DatanodeDetails dn3 = DatanodeDetails.newBuilder(protoBuilder.build()).build();
     assertEquals(HDDSVersion.SOFTWARE_VERSION,
         HDDSVersion.deserialize(dn3.getCurrentVersion()));

hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipeline.java

@@ -47,13 +47,13 @@ public void protoIncludesNewPortsOnlyForV1() throws IOException {
     Pipeline subject = MockPipeline.createPipeline(3);
 
     HddsProtos.Pipeline proto =
-        subject.getProtobufMessage(DEFAULT_VERSION.serialize());
+        subject.getProtobufMessage(DEFAULT_VERSION);
     for (HddsProtos.DatanodeDetailsProto dn : proto.getMembersList()) {
       assertPorts(dn, V0_PORTS);
     }
 
     HddsProtos.Pipeline protoV1 = subject.getProtobufMessage(
-        VERSION_HANDLES_UNKNOWN_DN_PORTS.serialize());
+        VERSION_HANDLES_UNKNOWN_DN_PORTS);
     for (HddsProtos.DatanodeDetailsProto dn : protoV1.getMembersList()) {
       assertPorts(dn, ALL_PORTS);
     }
@@ -64,14 +64,14 @@ public void getProtobufMessageEC() throws IOException {
     Pipeline subject = MockPipeline.createPipeline(3);
 
     //when EC config is empty/null
-    HddsProtos.Pipeline protobufMessage = subject.getProtobufMessage(1);
+    HddsProtos.Pipeline protobufMessage = subject.getProtobufMessage(VERSION_HANDLES_UNKNOWN_DN_PORTS);
     assertEquals(0, protobufMessage.getEcReplicationConfig().getData());
 
 
     //when EC config is NOT empty
     subject = MockPipeline.createEcPipeline();
 
-    protobufMessage = subject.getProtobufMessage(1);
+    protobufMessage = subject.getProtobufMessage(VERSION_HANDLES_UNKNOWN_DN_PORTS);
     assertEquals(3, protobufMessage.getEcReplicationConfig().getData());
     assertEquals(2, protobufMessage.getEcReplicationConfig().getParity());
 
@@ -80,7 +80,7 @@ public void getProtobufMessageEC() throws IOException {
   @Test
   public void testReplicaIndexesSerialisedCorrectly() {
     Pipeline pipeline = MockPipeline.createEcPipeline();
-    HddsProtos.Pipeline protobufMessage = pipeline.getProtobufMessage(1);
+    HddsProtos.Pipeline protobufMessage = pipeline.getProtobufMessage(VERSION_HANDLES_UNKNOWN_DN_PORTS);
     Pipeline reloadedPipeline = Pipeline.getFromProtobuf(protobufMessage);
 
     for (DatanodeDetails dn : pipeline.getNodes()) {

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

@@ -27,6 +27,7 @@
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.DiskBalancerConfigurationProto;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.DiskBalancerRunningStatus;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeOperationalState;
+import org.apache.hadoop.ozone.ClientVersion;
 import org.apache.hadoop.ozone.container.common.statemachine.DatanodeStateMachine;
 import org.apache.hadoop.ozone.container.ozoneimpl.OzoneContainer;
 import org.slf4j.Logger;
@@ -70,7 +71,7 @@ public DatanodeDiskBalancerInfoProto getDiskBalancerInfo(GetDiskBalancerInfoRequ
     DatanodeDetails datanodeDetails = datanodeStateMachine.getDatanodeDetails();
 
     return DatanodeDiskBalancerInfoProto.newBuilder()
-        .setNode(datanodeDetails.toProto(request.getClientVersion()))
+        .setNode(datanodeDetails.toProto(ClientVersion.deserialize(request.getClientVersion())))
         .setCurrentVolumeDensitySum(info.getVolumeDataDensity())
         .setDiskBalancerConf(DiskBalancerConfigurationProto.newBuilder()
                 .setThreshold(info.getThreshold())

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/container/common/helpers/MoveDataNodePair.java

@@ -35,7 +35,7 @@ public class MoveDataNodePair {
   private static final Codec<MoveDataNodePair> CODEC = new DelegatedCodec<>(
       Proto2Codec.get(MoveDataNodePairProto.getDefaultInstance()),
       MoveDataNodePair::getFromProtobuf,
-      pair -> pair.getProtobufMessage(ClientVersion.CURRENT.serialize()),
+      pair -> pair.getProtobufMessage(ClientVersion.CURRENT),
       MoveDataNodePair.class,
       DelegatedCodec.CopyType.SHALLOW);
 
@@ -66,7 +66,7 @@ public DatanodeDetails getSrc() {
     return src;
   }
 
-  public MoveDataNodePairProto getProtobufMessage(int clientVersion) {
+  public MoveDataNodePairProto getProtobufMessage(ClientVersion clientVersion) {
     return MoveDataNodePairProto.newBuilder()
         .setSrc(src.toProto(clientVersion))
         .setTgt(tgt.toProto(clientVersion))

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/net/InnerNodeImpl.java

@@ -31,6 +31,7 @@
 import java.util.Map;
 import java.util.Objects;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.ozone.ClientVersion;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -460,7 +461,7 @@ public Node getLeaf(int leafIndex, List<String> excludedScopes,
 
   @Override
   public HddsProtos.NetworkNode toProtobuf(
-      int clientVersion) {
+      ClientVersion clientVersion) {
 
     HddsProtos.InnerNode.Builder innerNode =
         HddsProtos.InnerNode.newBuilder()

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/protocol/StorageContainerLocationProtocol.java

@@ -46,6 +46,7 @@
 import org.apache.hadoop.hdds.scm.container.ReplicationManagerReport;
 import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline;
 import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
+import org.apache.hadoop.ozone.ClientVersion;
 import org.apache.hadoop.ozone.upgrade.UpgradeFinalization.StatusAndMessages;
 import org.apache.hadoop.security.KerberosInfo;
 import org.apache.hadoop.security.token.Token;
@@ -122,7 +123,7 @@ ContainerWithPipeline getContainerWithPipeline(long containerID)
    * @throws IOException
    */
   List<HddsProtos.SCMContainerReplicaProto> getContainerReplicas(
-      long containerId, int clientVersion) throws IOException;
+      long containerId, ClientVersion clientVersion) throws IOException;
 
   /**
    * Ask SCM the location of a batch of containers. SCM responds with a group of
@@ -281,7 +282,7 @@ ContainerListResult listContainer(long startContainerID,
    */
   List<HddsProtos.Node> queryNode(HddsProtos.NodeOperationalState opState,
       HddsProtos.NodeState state, HddsProtos.QueryScope queryScope,
-      String poolName, int clientVersion) throws IOException;
+      String poolName, ClientVersion clientVersion) throws IOException;
 
   HddsProtos.Node queryNode(UUID uuid) throws IOException;
 
@@ -495,7 +496,7 @@ StartContainerBalancerResponseProto startContainerBalancer(
    * @see org.apache.hadoop.ozone.ClientVersion
    */
   List<HddsProtos.DatanodeUsageInfoProto> getDatanodeUsageInfo(
-      String address, String uuid, int clientVersion) throws IOException;
+      String address, String uuid, ClientVersion clientVersion) throws IOException;
 
   /**
    * Get usage information of most or least used datanodes.
@@ -509,7 +510,7 @@ List<HddsProtos.DatanodeUsageInfoProto> getDatanodeUsageInfo(
    * @see org.apache.hadoop.ozone.ClientVersion
    */
   List<HddsProtos.DatanodeUsageInfoProto> getDatanodeUsageInfo(
-      boolean mostUsed, int count, int clientVersion) throws IOException;
+      boolean mostUsed, int count, ClientVersion clientVersion) throws IOException;
 
   @Deprecated
   StatusAndMessages finalizeScmUpgrade(String upgradeClientID)

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/protocolPB/StorageContainerLocationProtocolClientSideTranslatorPB.java

@@ -341,7 +341,7 @@ public ContainerWithPipeline getContainerWithPipeline(long containerID)
    */
   @Override
   public List<HddsProtos.SCMContainerReplicaProto> getContainerReplicas(
-      long containerID, int clientVersion) throws IOException {
+      long containerID, ClientVersion clientVersion) throws IOException {
     Preconditions.checkState(containerID >= 0,
         "Container ID cannot be negative");
 
@@ -563,7 +563,7 @@ public Map<String, List<ContainerID>> getContainersOnDecomNode(DatanodeDetails d
   public List<HddsProtos.Node> queryNode(
       HddsProtos.NodeOperationalState opState, HddsProtos.NodeState
       nodeState, HddsProtos.QueryScope queryScope, String poolName,
-      int clientVersion) throws IOException {
+      ClientVersion clientVersion) throws IOException {
     // TODO : We support only cluster wide query right now. So ignoring checking
     // queryScope and poolName
     NodeQueryRequestProto.Builder builder = NodeQueryRequestProto.newBuilder()
@@ -1135,7 +1135,7 @@ public ContainerBalancerStatusInfoResponseProto getContainerBalancerStatusInfo()
    */
   @Override
   public List<HddsProtos.DatanodeUsageInfoProto> getDatanodeUsageInfo(
-      String address, String uuid, int clientVersion) throws IOException {
+      String address, String uuid, ClientVersion clientVersion) throws IOException {
 
     DatanodeUsageInfoRequestProto request =
         DatanodeUsageInfoRequestProto.newBuilder()
@@ -1161,7 +1161,7 @@ public List<HddsProtos.DatanodeUsageInfoProto> getDatanodeUsageInfo(
    */
   @Override
   public List<HddsProtos.DatanodeUsageInfoProto> getDatanodeUsageInfo(
-      boolean mostUsed, int count, int clientVersion) throws IOException {
+      boolean mostUsed, int count, ClientVersion clientVersion) throws IOException {
     DatanodeUsageInfoRequestProto request =
         DatanodeUsageInfoRequestProto.newBuilder()
             .setMostUsed(mostUsed)

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

@@ -22,6 +22,7 @@
 import org.apache.hadoop.hdds.protocol.DatanodeID;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.DatanodeUsageInfoProto;
 import org.apache.hadoop.hdds.scm.container.placement.metrics.SCMNodeStat;
+import org.apache.hadoop.ozone.ClientVersion;
 
 /**
  * Bundles datanode details with usage statistics.
@@ -221,11 +222,11 @@ public int hashCode() {
    *
    * @return Protobuf HddsProtos.DatanodeUsageInfo
    */
-  public DatanodeUsageInfoProto toProto(int clientVersion) {
+  public DatanodeUsageInfoProto toProto(ClientVersion clientVersion) {
     return toProtoBuilder(clientVersion).build();
   }
 
-  private DatanodeUsageInfoProto.Builder toProtoBuilder(int clientVersion) {
+  private DatanodeUsageInfoProto.Builder toProtoBuilder(ClientVersion clientVersion) {
     DatanodeUsageInfoProto.Builder builder =
         DatanodeUsageInfoProto.newBuilder();
 

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java

@@ -268,7 +268,7 @@ private void checkIfPipelineCreationIsAllowed(
   private void addPipelineToManager(Pipeline pipeline)
       throws IOException {
     HddsProtos.Pipeline pipelineProto = pipeline.getProtobufMessage(
-        ClientVersion.CURRENT.serialize());
+        ClientVersion.CURRENT);
     acquireWriteLock();
     try {
       stateManager.addPipeline(pipelineProto);