HDDS-77. Key replication factor and type should be stored per key by Ozone Manager. Contributed by Mukul Kumar Singh.

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/ksm/helpers/KsmKeyInfo.java

@@ -18,6 +18,7 @@
 package org.apache.hadoop.ozone.ksm.helpers;
 
 import com.google.common.base.Preconditions;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 import org.apache.hadoop.ozone.protocol.proto.KeySpaceManagerProtocolProtos.KeyInfo;
 import org.apache.hadoop.util.Time;
 
@@ -39,10 +40,13 @@ public final class KsmKeyInfo {
   private List<KsmKeyLocationInfoGroup> keyLocationVersions;
   private final long creationTime;
   private long modificationTime;
+  private HddsProtos.ReplicationType type;
+  private HddsProtos.ReplicationFactor factor;
 
   private KsmKeyInfo(String volumeName, String bucketName, String keyName,
       List<KsmKeyLocationInfoGroup> versions, long dataSize,
-      long creationTime, long modificationTime) {
+      long creationTime, long modificationTime, HddsProtos.ReplicationType type,
+      HddsProtos.ReplicationFactor factor) {
     this.volumeName = volumeName;
     this.bucketName = bucketName;
     this.keyName = keyName;
@@ -61,6 +65,8 @@ private KsmKeyInfo(String volumeName, String bucketName, String keyName,
     this.keyLocationVersions = versions;
     this.creationTime = creationTime;
     this.modificationTime = modificationTime;
+    this.factor = factor;
+    this.type = type;
   }
 
   public String getVolumeName() {
@@ -71,6 +77,14 @@ public String getBucketName() {
     return bucketName;
   }
 
+  public HddsProtos.ReplicationType getType() {
+    return type;
+  }
+
+  public HddsProtos.ReplicationFactor getFactor() {
+    return factor;
+  }
+
   public String getKeyName() {
     return keyName;
   }
@@ -170,6 +184,8 @@ public static class Builder {
     private List<KsmKeyLocationInfoGroup> ksmKeyLocationInfoGroups;
     private long creationTime;
     private long modificationTime;
+    private HddsProtos.ReplicationType type;
+    private HddsProtos.ReplicationFactor factor;
 
     public Builder setVolumeName(String volume) {
       this.volumeName = volume;
@@ -207,10 +223,20 @@ public Builder setModificationTime(long mTime) {
       return this;
     }
 
+    public Builder setReplicationFactor(HddsProtos.ReplicationFactor factor) {
+      this.factor = factor;
+      return this;
+    }
+
+    public Builder setReplicationType(HddsProtos.ReplicationType type) {
+      this.type = type;
+      return this;
+    }
+
     public KsmKeyInfo build() {
       return new KsmKeyInfo(
           volumeName, bucketName, keyName, ksmKeyLocationInfoGroups,
-          dataSize, creationTime, modificationTime);
+          dataSize, creationTime, modificationTime, type, factor);
     }
   }
 
@@ -222,6 +248,8 @@ public KeyInfo getProtobuf() {
         .setBucketName(bucketName)
         .setKeyName(keyName)
         .setDataSize(dataSize)
+        .setFactor(factor)
+        .setType(type)
         .addAllKeyLocationList(keyLocationVersions.stream()
             .map(KsmKeyLocationInfoGroup::getProtobuf)
             .collect(Collectors.toList()))
@@ -241,7 +269,9 @@ public static KsmKeyInfo getFromProtobuf(KeyInfo keyInfo) {
             .collect(Collectors.toList()),
         keyInfo.getDataSize(),
         keyInfo.getCreationTime(),
-        keyInfo.getModificationTime());
+        keyInfo.getModificationTime(),
+        keyInfo.getType(),
+        keyInfo.getFactor());
   }
 
 }

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/ksm/protocolPB/KeySpaceManagerProtocolClientSideTranslatorPB.java

@@ -560,8 +560,6 @@ public KsmKeyLocationInfo allocateBlock(KsmKeyArgs args, int clientID)
         .setVolumeName(args.getVolumeName())
         .setBucketName(args.getBucketName())
         .setKeyName(args.getKeyName())
-        .setFactor(args.getFactor())
-        .setType(args.getType())
         .setDataSize(args.getDataSize()).build();
     req.setKeyArgs(keyArgs);
     req.setClientID(clientID);

hadoop-ozone/common/src/main/proto/KeySpaceManagerProtocol.proto

@@ -249,10 +249,12 @@ message KeyInfo {
     required string bucketName = 2;
     required string keyName = 3;
     required uint64 dataSize = 4;
-    repeated KeyLocationList keyLocationList = 5;
+    required hadoop.hdds.ReplicationType type = 5;
-    required uint64 creationTime = 6;
+    required hadoop.hdds.ReplicationFactor factor = 6;
-    required uint64 modificationTime = 7;
+    repeated KeyLocationList keyLocationList = 7;
-    optional uint64 latestVersion = 8;
+    required uint64 creationTime = 8;
+    required uint64 modificationTime = 9;
+    optional uint64 latestVersion = 10;
 }
 
 message LocateKeyRequest {

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/ksm/KeyManagerImpl.java

@@ -173,18 +173,6 @@ public KsmKeyLocationInfo allocateBlock(KsmKeyArgs args, int clientID)
     String volumeName = args.getVolumeName();
     String bucketName = args.getBucketName();
     String keyName = args.getKeyName();
-    ReplicationFactor factor = args.getFactor();
-    ReplicationType type = args.getType();
-
-    // If user does not specify a replication strategy or
-    // replication factor, KSM will use defaults.
-    if(factor == null) {
-      factor = useRatis ? ReplicationFactor.THREE: ReplicationFactor.ONE;
-    }
-
-    if(type == null) {
-      type = useRatis ? ReplicationType.RATIS : ReplicationType.STAND_ALONE;
-    }
 
     try {
       validateBucket(volumeName, bucketName);
@@ -198,10 +186,11 @@ public KsmKeyLocationInfo allocateBlock(KsmKeyArgs args, int clientID)
         throw new KSMException("Open Key not found",
             KSMException.ResultCodes.FAILED_KEY_NOT_FOUND);
       }
-      AllocatedBlock allocatedBlock =
-          scmBlockClient.allocateBlock(scmBlockSize, type, factor, ksmId);
       KsmKeyInfo keyInfo =
           KsmKeyInfo.getFromProtobuf(KeyInfo.parseFrom(keyData));
+      AllocatedBlock allocatedBlock =
+          scmBlockClient.allocateBlock(scmBlockSize, keyInfo.getType(),
+              keyInfo.getFactor(), ksmId);
       KsmKeyLocationInfo info = new KsmKeyLocationInfo.Builder()
           .setBlockID(allocatedBlock.getBlockID())
           .setShouldCreateContainer(allocatedBlock.getCreateContainer())
@@ -293,6 +282,8 @@ public OpenKeySession openKey(KsmKeyArgs args) throws IOException {
             .setCreationTime(currentTime)
             .setModificationTime(currentTime)
             .setDataSize(size)
+            .setReplicationType(type)
+            .setReplicationFactor(factor)
             .build();
         openVersion = 0;
       }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/KeySpaceManagerProtocolServerSideTranslatorPB.java

@@ -527,16 +527,10 @@ public AllocateBlockResponse allocateBlock(RpcController controller,
         AllocateBlockResponse.newBuilder();
     try {
       KeyArgs keyArgs = request.getKeyArgs();
-      HddsProtos.ReplicationType type =
-          keyArgs.hasType()? keyArgs.getType() : null;
-      HddsProtos.ReplicationFactor factor =
-          keyArgs.hasFactor()? keyArgs.getFactor() : null;
       KsmKeyArgs ksmKeyArgs = new KsmKeyArgs.Builder()
           .setVolumeName(keyArgs.getVolumeName())
           .setBucketName(keyArgs.getBucketName())
           .setKeyName(keyArgs.getKeyName())
-          .setType(type)
-          .setFactor(factor)
           .build();
       int id = request.getClientID();
       KsmKeyLocationInfo newLocation = impl.allocateBlock(ksmKeyArgs, id);