From 132a547dea4081948c39c149c59d6453003fa277 Mon Sep 17 00:00:00 2001
From: Mukul Kumar Singh <msingh@apache.org>
Date: Mon, 21 May 2018 22:57:08 +0530
Subject: [PATCH] HDDS-71. Send ContainerType to Datanode during container
 creation. Contributed by Bharat Viswanadham.

---
 .../scm/storage/ContainerProtocolCalls.java   |  2 ++
 .../proto/DatanodeContainerProtocol.proto     |  6 ++++
 .../common/helpers/ContainerData.java         | 36 +++++++++++++++++++
 .../common/impl/ContainerManagerImpl.java     |  7 ++++
 4 files changed, 51 insertions(+)

diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java
index 5fbf373d25b..d3af0836dea 100644
--- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java
@@ -243,6 +243,8 @@ public final class ContainerProtocolCalls  {
     ContainerProtos.ContainerData.Builder containerData = ContainerProtos
         .ContainerData.newBuilder();
     containerData.setContainerID(containerID);
+    containerData.setContainerType(ContainerProtos.ContainerType
+        .KeyValueContainer);
     createRequest.setContainerData(containerData.build());
 
     String id = client.getPipeline().getLeader().getUuidString();
diff --git a/hadoop-hdds/common/src/main/proto/DatanodeContainerProtocol.proto b/hadoop-hdds/common/src/main/proto/DatanodeContainerProtocol.proto
index 3479866d250..e7e5b2b6929 100644
--- a/hadoop-hdds/common/src/main/proto/DatanodeContainerProtocol.proto
+++ b/hadoop-hdds/common/src/main/proto/DatanodeContainerProtocol.proto
@@ -225,6 +225,12 @@ message ContainerData {
   optional int64 size = 7;
   optional int64 keyCount = 8;
   optional ContainerLifeCycleState state = 9 [default = OPEN];
+  optional ContainerType containerType = 10 [default = KeyValueContainer];
+  optional string containerDBType = 11;
+}
+
+enum ContainerType {
+  KeyValueContainer = 1;
 }
 
 
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ContainerData.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ContainerData.java
index 63111c814fe..2a079b0af01 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ContainerData.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ContainerData.java
@@ -21,6 +21,8 @@ package org.apache.hadoop.ozone.container.common.helpers;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hdds.scm.ScmConfigKeys;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos
+    .ContainerType;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos
     .ContainerLifeCycleState;
 import org.apache.hadoop.ozone.OzoneConsts;
@@ -47,6 +49,8 @@ public class ContainerData {
   private long maxSize;
   private long containerID;
   private ContainerLifeCycleState state;
+  private ContainerType containerType;
+  private String containerDBType;
 
   /**
    * Constructs a  ContainerData Object.
@@ -99,9 +103,26 @@ public class ContainerData {
     if (protoData.hasSize()) {
       data.setMaxSize(protoData.getSize());
     }
+
+    if(protoData.hasContainerType()) {
+      data.setContainerType(protoData.getContainerType());
+    }
+
+    if(protoData.hasContainerDBType()) {
+      data.setContainerDBType(protoData.getContainerDBType());
+    }
+
     return data;
   }
 
+  public String getContainerDBType() {
+    return containerDBType;
+  }
+
+  public void setContainerDBType(String containerDBType) {
+    this.containerDBType = containerDBType;
+  }
+
   /**
    * Returns a ProtoBuf Message from ContainerData.
    *
@@ -141,9 +162,24 @@ public class ContainerData {
       builder.setSize(this.getMaxSize());
     }
 
+    if(this.getContainerType() != null) {
+      builder.setContainerType(containerType);
+    }
+
+    if(this.getContainerDBType() != null) {
+      builder.setContainerDBType(containerDBType);
+    }
+
     return builder.build();
   }
 
+  public void setContainerType(ContainerType containerType) {
+    this.containerType = containerType;
+  }
+
+  public ContainerType getContainerType() {
+    return this.containerType;
+  }
   /**
    * Adds metadata.
    */
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/ContainerManagerImpl.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/ContainerManagerImpl.java
index 039b4c3293d..c443ace0854 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/ContainerManagerImpl.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/ContainerManagerImpl.java
@@ -42,6 +42,7 @@ import org.apache.hadoop.hdds.protocol.proto
 import org.apache.hadoop.hdds.protocol.proto
     .StorageContainerDatanodeProtocolProtos.StorageTypeProto;
 import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.ozone.OzoneConfigKeys;
 import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.hadoop.ozone.container.common.helpers.ContainerData;
 import org.apache.hadoop.ozone.container.common.helpers.ContainerUtils;
@@ -400,6 +401,12 @@ public class ContainerManagerImpl implements ContainerManager {
           .toString());
       containerData.setContainerPath(containerFile.toString());
 
+      if(containerData.getContainerDBType() == null) {
+        String impl = conf.getTrimmed(OzoneConfigKeys.OZONE_METADATA_STORE_IMPL,
+            OzoneConfigKeys.OZONE_METADATA_STORE_IMPL_DEFAULT);
+        containerData.setContainerDBType(impl);
+      }
+
       ContainerProtos.ContainerData protoData = containerData
           .getProtoBufMessage();
       protoData.writeDelimitedTo(dos);