Hdds 4575

hadoop-hdds/client/pom.xml

@@ -51,11 +51,6 @@ https://maven.apache.org/xsd/maven-4.0.0.xsd">
       <scope>test</scope>
     </dependency>
 
-    <dependency>
-      <groupId>io.netty</groupId>
-      <artifactId>netty-all</artifactId>
-    </dependency>
-
     <dependency>
       <groupId>org.apache.hadoop</groupId>
       <artifactId>hadoop-hdds-hadoop-dependency-test</artifactId>

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/OzoneClientConfig.java

@@ -21,6 +21,7 @@
 import org.apache.hadoop.hdds.conf.ConfigGroup;
 import org.apache.hadoop.hdds.conf.ConfigTag;
 import org.apache.hadoop.hdds.conf.ConfigType;
+import org.apache.hadoop.hdds.conf.PostConstruct;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChecksumType;
 import org.apache.hadoop.ozone.OzoneConfigKeys;
 
@@ -111,9 +112,7 @@ public class OzoneClientConfig {
       tags = ConfigTag.CLIENT)
   private boolean checksumVerify = true;
 
-  public OzoneClientConfig() {
-  }
-
+  @PostConstruct
   private void validate() {
     Preconditions.checkState(streamBufferSize > 0);
     Preconditions.checkState(streamBufferFlushSize > 0);

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientRatis.java

@@ -217,12 +217,12 @@ private CompletableFuture<RaftClientReply> sendRequestAsync(
             if (LOG.isDebugEnabled()) {
               LOG.debug("sendCommandAsync ReadOnly {}", message);
             }
-            return getClient().sendReadOnlyAsync(message);
+            return getClient().async().sendReadOnly(message);
           } else {
             if (LOG.isDebugEnabled()) {
               LOG.debug("sendCommandAsync {}", message);
             }
-            return getClient().sendAsync(message);
+            return getClient().async().send(message);
           }
 
         }
@@ -258,17 +258,17 @@ public XceiverClientReply watchForCommit(long index)
     }
     RaftClientReply reply;
     try {
-      CompletableFuture<RaftClientReply> replyFuture = getClient()
-          .sendWatchAsync(index, RaftProtos.ReplicationLevel.ALL_COMMITTED);
+      CompletableFuture<RaftClientReply> replyFuture = getClient().async()
+          .watch(index, RaftProtos.ReplicationLevel.ALL_COMMITTED);
       replyFuture.get();
     } catch (Exception e) {
       Throwable t = HddsClientUtils.checkForException(e);
       LOG.warn("3 way commit failed on pipeline {}", pipeline, e);
       if (t instanceof GroupMismatchException) {
         throw e;
       }
-      reply = getClient()
-          .sendWatchAsync(index, RaftProtos.ReplicationLevel.MAJORITY_COMMITTED)
+      reply = getClient().async()
+          .watch(index, RaftProtos.ReplicationLevel.MAJORITY_COMMITTED)
           .get();
       List<RaftProtos.CommitInfoProto> commitInfoProtoList =
           reply.getCommitInfos().stream()

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/ratis/RatisHelper.java

@@ -104,12 +104,18 @@ public static RaftPeerId toRaftPeerId(DatanodeDetails id) {
   }
 
   public static RaftPeer toRaftPeer(DatanodeDetails id) {
-    return new RaftPeer(toRaftPeerId(id), toRaftPeerAddressString(id));
+    return RaftPeer.newBuilder()
+        .setId(toRaftPeerId(id))
+        .setAddress(toRaftPeerAddressString(id))
+        .build();
   }
 
   public static RaftPeer toRaftPeer(DatanodeDetails id, int priority) {
-    return new RaftPeer(
-        toRaftPeerId(id), toRaftPeerAddressString(id), priority);
+    return RaftPeer.newBuilder()
+        .setId(toRaftPeerId(id))
+        .setAddress(toRaftPeerAddressString(id))
+        .setPriority(priority)
+        .build();
   }
 
   private static List<RaftPeer> toRaftPeers(Pipeline pipeline) {

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

@@ -273,6 +273,16 @@ public final class ScmConfigKeys {
   // able to send back a new list to the datanodes.
   public static final String OZONE_SCM_NAMES = "ozone.scm.names";
 
+  public static final String OZONE_SCM_INTERNAL_SERVICE_ID =
+      "ozone.scm.internal.service.id";
+
+  public static final String OZONE_SCM_SERVICE_IDS_KEY =
+      "ozone.scm.service.ids";
+  public static final String OZONE_SCM_NODES_KEY =
+      "ozone.scm.nodes";
+  public static final String OZONE_SCM_NODE_ID_KEY =
+      "ozone.scm.node.id";
+
   public static final int OZONE_SCM_DEFAULT_PORT =
       OZONE_SCM_DATANODE_PORT_DEFAULT;
   // The path where datanode ID is to be written to.
@@ -364,6 +374,83 @@ public final class ScmConfigKeys {
   public static final String HDDS_TRACING_ENABLED = "hdds.tracing.enabled";
   public static final boolean HDDS_TRACING_ENABLED_DEFAULT = false;
 
+  // SCM Ratis related
+  public static final String OZONE_SCM_HA_ENABLE_KEY
+      = "ozone.scm.ratis.enable";
+  public static final boolean OZONE_SCM_HA_ENABLE_DEFAULT
+      = false;
+  public static final String OZONE_SCM_RATIS_PORT_KEY
+      = "ozone.scm.ratis.port";
+  public static final int OZONE_SCM_RATIS_PORT_DEFAULT
+      = 9864;
+  public static final String OZONE_SCM_RATIS_RPC_TYPE_KEY
+      = "ozone.scm.ratis.rpc.type";
+  public static final String OZONE_SCM_RATIS_RPC_TYPE_DEFAULT
+      = "GRPC";
+
+  // SCM Ratis Log configurations
+  public static final String OZONE_SCM_RATIS_STORAGE_DIR
+      = "ozone.scm.ratis.storage.dir";
+  public static final String OZONE_SCM_RATIS_SEGMENT_SIZE_KEY
+      = "ozone.scm.ratis.segment.size";
+  public static final String OZONE_SCM_RATIS_SEGMENT_SIZE_DEFAULT
+      = "16KB";
+  public static final String OZONE_SCM_RATIS_SEGMENT_PREALLOCATED_SIZE_KEY
+      = "ozone.scm.ratis.segment.preallocated.size";
+  public static final String OZONE_SCM_RATIS_SEGMENT_PREALLOCATED_SIZE_DEFAULT
+      = "16KB";
+
+  // SCM Ratis Log Appender configurations
+  public static final String
+      OZONE_SCM_RATIS_LOG_APPENDER_QUEUE_NUM_ELEMENTS =
+      "ozone.scm.ratis.log.appender.queue.num-elements";
+  public static final int
+      OZONE_SCM_RATIS_LOG_APPENDER_QUEUE_NUM_ELEMENTS_DEFAULT = 1024;
+  public static final String OZONE_SCM_RATIS_LOG_APPENDER_QUEUE_BYTE_LIMIT =
+      "ozone.scm.ratis.log.appender.queue.byte-limit";
+  public static final String
+      OZONE_SCM_RATIS_LOG_APPENDER_QUEUE_BYTE_LIMIT_DEFAULT = "32MB";
+  public static final String OZONE_SCM_RATIS_LOG_PURGE_GAP =
+      "ozone.scm.ratis.log.purge.gap";
+  public static final int OZONE_SCM_RATIS_LOG_PURGE_GAP_DEFAULT = 1000000;
+
+  // SCM Ratis server configurations
+  public static final String OZONE_SCM_RATIS_SERVER_REQUEST_TIMEOUT_KEY
+      = "ozone.scm.ratis.server.request.timeout";
+  public static final TimeDuration
+      OZONE_SCM_RATIS_SERVER_REQUEST_TIMEOUT_DEFAULT
+      = TimeDuration.valueOf(3000, TimeUnit.MILLISECONDS);
+  public static final String
+      OZONE_SCM_RATIS_SERVER_RETRY_CACHE_TIMEOUT_KEY
+      = "ozone.scm.ratis.server.retry.cache.timeout";
+  public static final TimeDuration
+      OZONE_SCM_RATIS_SERVER_RETRY_CACHE_TIMEOUT_DEFAULT
+      = TimeDuration.valueOf(600000, TimeUnit.MILLISECONDS);
+  public static final String OZONE_SCM_RATIS_MINIMUM_TIMEOUT_KEY
+      = "ozone.scm.ratis.minimum.timeout";
+  public static final TimeDuration OZONE_SCM_RATIS_MINIMUM_TIMEOUT_DEFAULT
+      = TimeDuration.valueOf(1, TimeUnit.SECONDS);
+
+  // SCM Ratis Leader Election configurations
+  public static final String
+      OZONE_SCM_LEADER_ELECTION_MINIMUM_TIMEOUT_DURATION_KEY =
+      "ozone.scm.ratis.leader.election.minimum.timeout.duration";
+  public static final TimeDuration
+      OZONE_SCM_LEADER_ELECTION_MINIMUM_TIMEOUT_DURATION_DEFAULT =
+      TimeDuration.valueOf(1, TimeUnit.SECONDS);
+  public static final String OZONE_SCM_RATIS_SERVER_FAILURE_TIMEOUT_DURATION_KEY
+      = "ozone.scm.ratis.server.failure.timeout.duration";
+  public static final TimeDuration
+      OZONE_SCM_RATIS_SERVER_FAILURE_TIMEOUT_DURATION_DEFAULT
+      = TimeDuration.valueOf(120, TimeUnit.SECONDS);
+
+  // SCM Leader server role check interval
+  public static final String OZONE_SCM_RATIS_SERVER_ROLE_CHECK_INTERVAL_KEY
+      = "ozone.scm.ratis.server.role.check.interval";
+  public static final TimeDuration
+      OZONE_SCM_RATIS_SERVER_ROLE_CHECK_INTERVAL_DEFAULT
+      = TimeDuration.valueOf(15, TimeUnit.SECONDS);
+
   /**
    * Never constructed.
    */

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

@@ -18,20 +18,29 @@
 
 package org.apache.hadoop.hdds.scm;
 
+import java.util.ArrayList;
+import java.util.List;
+
 /**
  * ScmInfo wraps the result returned from SCM#getScmInfo which
  * contains clusterId and the SCM Id.
  */
 public final class ScmInfo {
   private String clusterId;
   private String scmId;
+  private List<String> peerRoles;
 
   /**
    * Builder for ScmInfo.
    */
   public static class Builder {
     private String clusterId;
     private String scmId;
+    private List<String> peerRoles;
+
+    public Builder() {
+      peerRoles = new ArrayList<>();
+    }
 
     /**
      * sets the cluster id.
@@ -53,14 +62,25 @@ public Builder setScmId(String id) {
       return this;
     }
 
+    /**
+     * Set peer address in Scm HA.
+     * @param roles ratis peer address in the format of [ip|hostname]:port
+     * @return  Builder for scmInfo
+     */
+    public Builder setRatisPeerRoles(List<String> roles) {
+      peerRoles.addAll(roles);
+      return this;
+    }
+
     public ScmInfo build() {
-      return new ScmInfo(clusterId, scmId);
+      return new ScmInfo(clusterId, scmId, peerRoles);
     }
   }
 
-  private ScmInfo(String clusterId, String scmId) {
+  private ScmInfo(String clusterId, String scmId, List<String> peerRoles) {
     this.clusterId = clusterId;
     this.scmId = scmId;
+    this.peerRoles = peerRoles;
   }
 
   /**
@@ -78,4 +98,12 @@ public String getClusterId() {
   public String getScmId() {
     return scmId;
   }
+
+  /**
+   * Gets the list of peer roles (currently address) in Scm HA.
+   * @return List of peer address
+   */
+  public List<String> getRatisPeerRoles() {
+    return peerRoles;
+  }
 }

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/client/ScmClient.java

@@ -246,5 +246,8 @@ Map<String, Pair<Boolean, String>> getSafeModeRuleStatuses()
    */
   boolean getReplicationManagerStatus() throws IOException;
 
-
+  /**
+   * returns the list of ratis peer roles. Currently only include peer address.
+   */
+  List<String> getScmRatisRoles() throws IOException;
 }

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

@@ -23,6 +23,7 @@
 import org.apache.commons.lang3.builder.CompareToBuilder;
 import org.apache.commons.lang3.builder.EqualsBuilder;
 import org.apache.commons.lang3.builder.HashCodeBuilder;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 
 /**
  * Container ID is an integer that is a value between 1..MAX_CONTAINER ID.
@@ -34,13 +35,14 @@ public final class ContainerID implements Comparable<ContainerID> {
 
   private final long id;
 
-  // TODO: make this private.
   /**
    * Constructs ContainerID.
    *
    * @param id int
    */
-  public ContainerID(long id) {
+  private ContainerID(long id) {
+    Preconditions.checkState(id > 0,
+        "Container ID should be a positive. %s.", id);
     this.id = id;
   }
 
@@ -49,9 +51,7 @@ public ContainerID(long id) {
    * @param containerID  long
    * @return ContainerID.
    */
-  public static ContainerID valueof(final long containerID) {
-    Preconditions.checkState(containerID > 0,
-        "Container ID should be a positive long. "+ containerID);
+  public static ContainerID valueOf(final long containerID) {
     return new ContainerID(containerID);
   }
 
@@ -60,14 +60,30 @@ public static ContainerID valueof(final long containerID) {
    *
    * @return int
    */
+  @Deprecated
+  /*
+   * Don't expose the int value.
+   */
   public long getId() {
     return id;
   }
 
+  /**
+   * Use proto message.
+   */
+  @Deprecated
   public byte[] getBytes() {
     return Longs.toByteArray(id);
   }
 
+  public HddsProtos.ContainerID getProtobuf() {
+    return HddsProtos.ContainerID.newBuilder().setId(id).build();
+  }
+
+  public static ContainerID getFromProtobuf(HddsProtos.ContainerID proto) {
+    return ContainerID.valueOf(proto.getId());
+  }
+
   @Override
   public boolean equals(final Object o) {
     if (this == o) {
@@ -81,22 +97,22 @@ public boolean equals(final Object o) {
     final ContainerID that = (ContainerID) o;
 
     return new EqualsBuilder()
-        .append(getId(), that.getId())
+        .append(id, that.id)
         .isEquals();
   }
 
   @Override
   public int hashCode() {
     return new HashCodeBuilder(61, 71)
-        .append(getId())
+        .append(id)
         .toHashCode();
   }
 
   @Override
   public int compareTo(final ContainerID that) {
     Preconditions.checkNotNull(that);
     return new CompareToBuilder()
-        .append(this.getId(), that.getId())
+        .append(this.id, that.id)
         .build();
   }
 

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

@@ -121,6 +121,11 @@ public static ContainerInfo fromProtobuf(HddsProtos.ContainerInfoProto info) {
         .build();
   }
 
+  /**
+   * This method is depricated, use {@code containerID()} which returns
+   * {@link ContainerID} object.
+   */
+  @Deprecated
   public long getContainerID() {
     return containerID;
   }
@@ -179,7 +184,7 @@ public void updateSequenceId(long sequenceID) {
   }
 
   public ContainerID containerID() {
-    return new ContainerID(getContainerID());
+    return ContainerID.valueOf(containerID);
   }
 
   /**

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

@@ -91,7 +91,7 @@ public static ExcludeList getFromProtoBuf(
       HddsProtos.ExcludeListProto excludeListProto) {
     ExcludeList excludeList = new ExcludeList();
     excludeListProto.getContainerIdsList().forEach(id -> {
-      excludeList.addConatinerId(ContainerID.valueof(id));
+      excludeList.addConatinerId(ContainerID.valueOf(id));
     });
     DatanodeDetails.Builder builder = DatanodeDetails.newBuilder();
     excludeListProto.getDatanodesList().forEach(dn -> {