hadoop-common-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jiten...@apache.org
Subject hadoop git commit: HDDS-550. Serialize ApplyTransaction calls per Container in ContainerStateMachine. Contributed by Shashikant Banerjee.
Date Fri, 12 Oct 2018 00:13:31 GMT
Repository: hadoop
Updated Branches:
  refs/heads/trunk 8d1981806 -> 0bc6d0484


HDDS-550. Serialize ApplyTransaction calls per Container in ContainerStateMachine. Contributed by Shashikant Banerjee.


Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/0bc6d048
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/0bc6d048
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/0bc6d048

Branch: refs/heads/trunk
Commit: 0bc6d0484a1c3652234b76e4f99e2ef99173796d
Parents: 8d19818
Author: Jitendra Pandey <jitendra@apache.org>
Authored: Thu Oct 11 16:59:59 2018 -0700
Committer: Jitendra Pandey <jitendra@apache.org>
Committed: Thu Oct 11 17:12:47 2018 -0700

----------------------------------------------------------------------
 .../hadoop/hdds/scm/XceiverClientGrpc.java      |   6 +-
 .../hadoop/hdds/scm/XceiverClientRatis.java     |   4 +-
 .../hdds/scm/storage/ChunkInputStream.java      |   2 +-
 .../hdds/scm/storage/ChunkOutputStream.java     |   3 +-
 hadoop-hdds/common/pom.xml                      |   4 +-
 .../apache/hadoop/hdds/scm/ScmConfigKeys.java   |   6 +-
 .../scm/storage/ContainerProtocolCalls.java     |   2 +-
 .../apache/hadoop/ozone/OzoneConfigKeys.java    |   6 +-
 .../org/apache/hadoop/utils/RocksDBStore.java   |   3 +-
 .../org/apache/hadoop/utils/db/RDBStore.java    |   3 +-
 .../main/java/org/apache/ratis/RatisHelper.java |  14 +-
 .../main/proto/DatanodeContainerProtocol.proto  |   1 -
 .../common/src/main/resources/ozone-default.xml |   8 +
 .../transport/server/GrpcXceiverService.java    |   2 +-
 .../transport/server/XceiverServerGrpc.java     |   8 +-
 .../server/ratis/ContainerStateMachine.java     | 440 +++++--------------
 .../server/ratis/XceiverServerRatis.java        |  15 +-
 .../container/keyvalue/helpers/ChunkUtils.java  |   2 +-
 .../keyvalue/helpers/SmallFileUtils.java        |   2 +-
 .../background/BlockDeletingService.java        |   2 +-
 .../replication/GrpcReplicationClient.java      |   6 +-
 .../replication/GrpcReplicationService.java     |   4 +-
 .../common/impl/TestHddsDispatcher.java         |   2 +-
 .../ozone/client/io/ChunkGroupOutputStream.java |   5 +
 .../rpc/TestCloseContainerHandlingByClient.java |   1 -
 .../ozone/container/ContainerTestHelper.java    |   3 +-
 .../common/impl/TestCloseContainerHandler.java  |   2 +-
 .../container/server/TestContainerServer.java   |   1 -
 .../server/TestContainerStateMachine.java       | 201 ---------
 .../genesis/BenchMarkDatanodeDispatcher.java    |   2 +-
 hadoop-project/pom.xml                          |   2 +-
 31 files changed, 192 insertions(+), 570 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hadoop/blob/0bc6d048/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java
index d353e7a..2f11872 100644
--- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java
+++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java
@@ -31,9 +31,9 @@ import org.apache.hadoop.hdds.scm.client.HddsClientUtils;
 import org.apache.hadoop.hdds.scm.container.common.helpers.Pipeline;
 import org.apache.hadoop.ozone.OzoneConfigKeys;
 import org.apache.hadoop.util.Time;
-import org.apache.ratis.shaded.io.grpc.ManagedChannel;
-import org.apache.ratis.shaded.io.grpc.netty.NettyChannelBuilder;
-import org.apache.ratis.shaded.io.grpc.stub.StreamObserver;
+import org.apache.ratis.thirdparty.io.grpc.ManagedChannel;
+import org.apache.ratis.thirdparty.io.grpc.netty.NettyChannelBuilder;
+import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

http://git-wip-us.apache.org/repos/asf/hadoop/blob/0bc6d048/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientRatis.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientRatis.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientRatis.java
index 0d301d9..4efe7ba 100644
--- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientRatis.java
+++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientRatis.java
@@ -21,7 +21,7 @@ package org.apache.hadoop.hdds.scm;
 import org.apache.hadoop.hdds.HddsUtils;
 import org.apache.hadoop.io.MultipleIOException;
 import org.apache.ratis.retry.RetryPolicy;
-import org.apache.ratis.shaded.com.google.protobuf
+import org.apache.ratis.thirdparty.com.google.protobuf
     .InvalidProtocolBufferException;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hdds.scm.client.HddsClientUtils;
@@ -39,7 +39,7 @@ import org.apache.ratis.protocol.RaftGroup;
 import org.apache.ratis.protocol.RaftPeer;
 import org.apache.ratis.rpc.RpcType;
 import org.apache.ratis.rpc.SupportedRpcType;
-import org.apache.ratis.shaded.com.google.protobuf.ByteString;
+import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
 import org.apache.ratis.util.CheckedBiConsumer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

http://git-wip-us.apache.org/repos/asf/hadoop/blob/0bc6d048/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java
index a483197..21b8974 100644
--- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java
+++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java
@@ -18,7 +18,7 @@
 
 package org.apache.hadoop.hdds.scm.storage;
 
-import org.apache.ratis.shaded.com.google.protobuf.ByteString;
+import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
 import org.apache.hadoop.fs.Seekable;
 import org.apache.hadoop.hdds.scm.XceiverClientManager;
 import org.apache.hadoop.hdds.scm.XceiverClientSpi;

http://git-wip-us.apache.org/repos/asf/hadoop/blob/0bc6d048/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkOutputStream.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkOutputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkOutputStream.java
index cc1ea8d..4547163 100644
--- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkOutputStream.java
+++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkOutputStream.java
@@ -18,8 +18,9 @@
 
 package org.apache.hadoop.hdds.scm.storage;
 
+
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
-import org.apache.ratis.shaded.com.google.protobuf.ByteString;
+import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
 import org.apache.commons.codec.digest.DigestUtils;
 import org.apache.hadoop.hdds.scm.XceiverClientManager;
 import org.apache.hadoop.hdds.scm.XceiverClientSpi;

http://git-wip-us.apache.org/repos/asf/hadoop/blob/0bc6d048/hadoop-hdds/common/pom.xml
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/pom.xml b/hadoop-hdds/common/pom.xml
index eea2264..bf2a6b9 100644
--- a/hadoop-hdds/common/pom.xml
+++ b/hadoop-hdds/common/pom.xml
@@ -172,10 +172,10 @@ http://maven.apache.org/xsd/maven-4.0.0.xsd">
             <phase>generate-sources</phase>
             <configuration>
               <tasks>
-                <replace token="com.google.protobuf" value="org.apache.ratis.shaded.com.google.protobuf"
+                <replace token="com.google.protobuf" value="org.apache.ratis.thirdparty.com.google.protobuf"
                   dir="target/generated-sources/java/org/apache/hadoop/hdds/protocol/datanode/proto">
                 </replace>
-                <replace token="io.grpc" value="org.apache.ratis.shaded.io.grpc"
+                <replace token="io.grpc" value="org.apache.ratis.thirdparty.io.grpc"
                   dir="target/generated-sources/java/org/apache/hadoop/hdds/protocol/datanode/proto">
                 </replace>
               </tasks>

http://git-wip-us.apache.org/repos/asf/hadoop/blob/0bc6d048/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfigKeys.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfigKeys.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfigKeys.java
index 63f5916..71d6e07 100644
--- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfigKeys.java
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfigKeys.java
@@ -19,7 +19,7 @@ package org.apache.hadoop.hdds.scm;
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
-import org.apache.ratis.shaded.proto.RaftProtos.ReplicationLevel;
+import org.apache.ratis.proto.RaftProtos.ReplicationLevel;
 import org.apache.ratis.util.TimeDuration;
 
 import java.util.concurrent.TimeUnit;
@@ -62,6 +62,10 @@ public final class ScmConfigKeys {
       = "dfs.container.ratis.replication.level";
   public static final ReplicationLevel
       DFS_CONTAINER_RATIS_REPLICATION_LEVEL_DEFAULT = ReplicationLevel.MAJORITY;
+  public static final String DFS_CONTAINER_RATIS_NUM_CONTAINER_OP_EXECUTORS_KEY
+      = "dfs.container.ratis.num.container.op.threads";
+  public static final int DFS_CONTAINER_RATIS_NUM_CONTAINER_OP_EXECUTORS_DEFAULT
+      = 10;
   public static final String DFS_CONTAINER_RATIS_SEGMENT_SIZE_KEY =
       "dfs.container.ratis.segment.size";
   public static final int DFS_CONTAINER_RATIS_SEGMENT_SIZE_DEFAULT =

http://git-wip-us.apache.org/repos/asf/hadoop/blob/0bc6d048/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java
----------------------------------------------------------------------
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 1df50b1..278b129 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
@@ -20,7 +20,7 @@ package org.apache.hadoop.hdds.scm.storage;
 
 import org.apache.hadoop.hdds.scm.container.common.helpers
     .BlockNotCommittedException;
-import org.apache.ratis.shaded.com.google.protobuf.ByteString;
+import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
 import org.apache.hadoop.hdds.scm.XceiverClientSpi;
 import org.apache.hadoop.hdds.scm.container.common.helpers
     .StorageContainerException;

http://git-wip-us.apache.org/repos/asf/hadoop/blob/0bc6d048/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java
index e8aa22c..d6a6bf7 100644
--- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java
@@ -24,7 +24,7 @@ import org.apache.hadoop.hdds.client.ReplicationFactor;
 import org.apache.hadoop.hdds.client.ReplicationType;
 import org.apache.hadoop.hdds.scm.ScmConfigKeys;
 
-import org.apache.ratis.shaded.proto.RaftProtos.ReplicationLevel;
+import org.apache.ratis.proto.RaftProtos.ReplicationLevel;
 import org.apache.ratis.util.TimeDuration;
 
 /**
@@ -220,6 +220,10 @@ public final class OzoneConfigKeys {
   public static final ReplicationLevel
       DFS_CONTAINER_RATIS_REPLICATION_LEVEL_DEFAULT
       = ScmConfigKeys.DFS_CONTAINER_RATIS_REPLICATION_LEVEL_DEFAULT;
+  public static final String DFS_CONTAINER_RATIS_NUM_CONTAINER_OP_EXECUTORS_KEY
+      = ScmConfigKeys.DFS_CONTAINER_RATIS_NUM_CONTAINER_OP_EXECUTORS_KEY;
+  public static final int DFS_CONTAINER_RATIS_NUM_CONTAINER_OP_EXECUTORS_DEFAULT
+      = ScmConfigKeys.DFS_CONTAINER_RATIS_NUM_CONTAINER_OP_EXECUTORS_DEFAULT;
   public static final String DFS_CONTAINER_RATIS_SEGMENT_SIZE_KEY
       = ScmConfigKeys.DFS_CONTAINER_RATIS_SEGMENT_SIZE_KEY;
   public static final int DFS_CONTAINER_RATIS_SEGMENT_SIZE_DEFAULT

http://git-wip-us.apache.org/repos/asf/hadoop/blob/0bc6d048/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/RocksDBStore.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/RocksDBStore.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/RocksDBStore.java
index 379d9e9..377153a 100644
--- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/RocksDBStore.java
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/RocksDBStore.java
@@ -22,7 +22,8 @@ import com.google.common.base.Preconditions;
 import org.apache.commons.io.FileUtils;
 import org.apache.commons.lang3.tuple.ImmutablePair;
 import org.apache.hadoop.metrics2.util.MBeans;
-import org.apache.ratis.shaded.com.google.common.annotations.VisibleForTesting;
+import org.apache.ratis.thirdparty.com.google.common.annotations.
+    VisibleForTesting;
 import org.rocksdb.DbPath;
 import org.rocksdb.Options;
 import org.rocksdb.RocksDB;

http://git-wip-us.apache.org/repos/asf/hadoop/blob/0bc6d048/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/RDBStore.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/RDBStore.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/RDBStore.java
index 5078b3e..cdee10b 100644
--- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/RDBStore.java
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/RDBStore.java
@@ -23,7 +23,8 @@ import com.google.common.base.Preconditions;
 import org.apache.hadoop.hdfs.DFSUtil;
 import org.apache.hadoop.metrics2.util.MBeans;
 import org.apache.hadoop.utils.RocksDBStoreMBean;
-import org.apache.ratis.shaded.com.google.common.annotations.VisibleForTesting;
+import org.apache.ratis.thirdparty.com.google.common.annotations.
+    VisibleForTesting;
 import org.rocksdb.ColumnFamilyDescriptor;
 import org.rocksdb.ColumnFamilyHandle;
 

http://git-wip-us.apache.org/repos/asf/hadoop/blob/0bc6d048/hadoop-hdds/common/src/main/java/org/apache/ratis/RatisHelper.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/java/org/apache/ratis/RatisHelper.java b/hadoop-hdds/common/src/main/java/org/apache/ratis/RatisHelper.java
index 04bfeb2..2dbe2e6 100644
--- a/hadoop-hdds/common/src/main/java/org/apache/ratis/RatisHelper.java
+++ b/hadoop-hdds/common/src/main/java/org/apache/ratis/RatisHelper.java
@@ -32,8 +32,8 @@ import org.apache.ratis.protocol.RaftPeerId;
 import org.apache.ratis.retry.RetryPolicies;
 import org.apache.ratis.retry.RetryPolicy;
 import org.apache.ratis.rpc.RpcType;
-import org.apache.ratis.shaded.com.google.protobuf.ByteString;
-import org.apache.ratis.shaded.proto.RaftProtos;
+import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
+import org.apache.ratis.proto.RaftProtos;
 import org.apache.ratis.util.Preconditions;
 import org.apache.ratis.util.SizeInBytes;
 import org.apache.ratis.util.TimeDuration;
@@ -103,7 +103,7 @@ public interface RatisHelper {
   RaftGroupId DUMMY_GROUP_ID =
       RaftGroupId.valueOf(ByteString.copyFromUtf8("AOzoneRatisGroup"));
 
-  RaftGroup EMPTY_GROUP = new RaftGroup(DUMMY_GROUP_ID,
+  RaftGroup EMPTY_GROUP = RaftGroup.valueOf(DUMMY_GROUP_ID,
       Collections.emptyList());
 
   static RaftGroup emptyRaftGroup() {
@@ -112,7 +112,7 @@ public interface RatisHelper {
 
   static RaftGroup newRaftGroup(Collection<RaftPeer> peers) {
     return peers.isEmpty()? emptyRaftGroup()
-        : new RaftGroup(DUMMY_GROUP_ID, peers);
+        : RaftGroup.valueOf(DUMMY_GROUP_ID, peers);
   }
 
   static RaftGroup newRaftGroup(RaftGroupId groupId,
@@ -120,12 +120,12 @@ public interface RatisHelper {
     final List<RaftPeer> newPeers = peers.stream()
         .map(RatisHelper::toRaftPeer)
         .collect(Collectors.toList());
-    return peers.isEmpty() ? new RaftGroup(groupId, Collections.emptyList())
-        : new RaftGroup(groupId, newPeers);
+    return peers.isEmpty() ? RaftGroup.valueOf(groupId, Collections.emptyList())
+        : RaftGroup.valueOf(groupId, newPeers);
   }
 
   static RaftGroup newRaftGroup(Pipeline pipeline) {
-    return new RaftGroup(pipeline.getId().getRaftGroupID(),
+    return RaftGroup.valueOf(pipeline.getId().getRaftGroupID(),
         toRaftPeers(pipeline));
   }
 

http://git-wip-us.apache.org/repos/asf/hadoop/blob/0bc6d048/hadoop-hdds/common/src/main/proto/DatanodeContainerProtocol.proto
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/proto/DatanodeContainerProtocol.proto b/hadoop-hdds/common/src/main/proto/DatanodeContainerProtocol.proto
index 4567750..662df8f 100644
--- a/hadoop-hdds/common/src/main/proto/DatanodeContainerProtocol.proto
+++ b/hadoop-hdds/common/src/main/proto/DatanodeContainerProtocol.proto
@@ -197,7 +197,6 @@ message ContainerCommandRequestProto {
 
   optional   PutSmallFileRequestProto putSmallFile = 19;
   optional   GetSmallFileRequestProto getSmallFile = 20;
-
   optional   GetCommittedBlockLengthRequestProto getCommittedBlockLength = 21;
 }
 

http://git-wip-us.apache.org/repos/asf/hadoop/blob/0bc6d048/hadoop-hdds/common/src/main/resources/ozone-default.xml
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/resources/ozone-default.xml b/hadoop-hdds/common/src/main/resources/ozone-default.xml
index ef54ed2..0fde6bb 100644
--- a/hadoop-hdds/common/src/main/resources/ozone-default.xml
+++ b/hadoop-hdds/common/src/main/resources/ozone-default.xml
@@ -136,6 +136,14 @@
     </description>
   </property>
   <property>
+    <name>dfs.container.ratis.num.container.op.executors</name>
+    <value>10</value>
+    <tag>OZONE, RATIS, PERFORMANCE</tag>
+    <description>Number of executors that will be used by Ratis to execute
+      container ops.(10 by default).
+    </description>
+  </property>
+  <property>
     <name>dfs.container.ratis.segment.size</name>
     <value>1073741824</value>
     <tag>OZONE, RATIS, PERFORMANCE</tag>

http://git-wip-us.apache.org/repos/asf/hadoop/blob/0bc6d048/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/GrpcXceiverService.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/GrpcXceiverService.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/GrpcXceiverService.java
index db4a86a..5fc3661 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/GrpcXceiverService.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/GrpcXceiverService.java
@@ -24,7 +24,7 @@ import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos
 import org.apache.hadoop.hdds.protocol.datanode.proto
     .XceiverClientProtocolServiceGrpc;
 import org.apache.hadoop.ozone.container.common.interfaces.ContainerDispatcher;
-import org.apache.ratis.shaded.io.grpc.stub.StreamObserver;
+import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

http://git-wip-us.apache.org/repos/asf/hadoop/blob/0bc6d048/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/XceiverServerGrpc.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/XceiverServerGrpc.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/XceiverServerGrpc.java
index c51da98..8ebfe49 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/XceiverServerGrpc.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/XceiverServerGrpc.java
@@ -33,10 +33,10 @@ import org.apache.hadoop.hdds.scm.container.common.helpers.
 import org.apache.hadoop.ozone.OzoneConfigKeys;
 import org.apache.hadoop.ozone.container.common.interfaces.ContainerDispatcher;
 
-import org.apache.ratis.shaded.io.grpc.BindableService;
-import org.apache.ratis.shaded.io.grpc.Server;
-import org.apache.ratis.shaded.io.grpc.ServerBuilder;
-import org.apache.ratis.shaded.io.grpc.netty.NettyServerBuilder;
+import org.apache.ratis.thirdparty.io.grpc.BindableService;
+import org.apache.ratis.thirdparty.io.grpc.Server;
+import org.apache.ratis.thirdparty.io.grpc.ServerBuilder;
+import org.apache.ratis.thirdparty.io.grpc.netty.NettyServerBuilder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

http://git-wip-us.apache.org/repos/asf/hadoop/blob/0bc6d048/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java
index f07c95f..7b7be91 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java
@@ -18,7 +18,6 @@
 
 package org.apache.hadoop.ozone.container.common.transport.server.ratis;
 
-import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import org.apache.hadoop.hdds.HddsUtils;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
@@ -26,7 +25,7 @@ import org.apache.hadoop.ozone.container.common.helpers.BlockData;
 import org.apache.ratis.protocol.RaftGroup;
 import org.apache.ratis.protocol.RaftGroupId;
 import org.apache.ratis.server.RaftServer;
-import org.apache.ratis.shaded.com.google.protobuf
+import org.apache.ratis.thirdparty.com.google.protobuf
     .InvalidProtocolBufferException;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Type;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Stage;
@@ -44,10 +43,10 @@ import org.apache.hadoop.ozone.container.common.interfaces.ContainerDispatcher;
 import org.apache.ratis.protocol.Message;
 import org.apache.ratis.protocol.RaftClientRequest;
 import org.apache.ratis.server.storage.RaftStorage;
-import org.apache.ratis.shaded.com.google.protobuf.ByteString;
-import org.apache.ratis.shaded.proto.RaftProtos.RoleInfoProto;
-import org.apache.ratis.shaded.proto.RaftProtos.LogEntryProto;
-import org.apache.ratis.shaded.proto.RaftProtos.SMLogEntryProto;
+import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
+import org.apache.ratis.proto.RaftProtos.RoleInfoProto;
+import org.apache.ratis.proto.RaftProtos.LogEntryProto;
+import org.apache.ratis.proto.RaftProtos.SMLogEntryProto;
 import org.apache.ratis.statemachine.StateMachineStorage;
 import org.apache.ratis.statemachine.TransactionContext;
 import org.apache.ratis.statemachine.impl.BaseStateMachine;
@@ -57,12 +56,12 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ExecutorService;
 import java.util.stream.Collectors;
 
 /** A {@link org.apache.ratis.statemachine.StateMachine} for containers.
@@ -98,44 +97,43 @@ import java.util.stream.Collectors;
  *
  * 2) Write chunk commit operation is executed after write chunk state machine
  * operation. This will ensure that commit operation is sync'd with the state
- * machine operation.
- *
- * Synchronization between {@link #writeStateMachineData} and
- * {@link #applyTransaction} need to be enforced in the StateMachine
- * implementation. For example, synchronization between writeChunk and
+ * machine operation.For example, synchronization between writeChunk and
  * createContainer in {@link ContainerStateMachine}.
- *
- * PutBlock is synchronized with WriteChunk operations, PutBlock for a block is
- * executed only after all the WriteChunk preceding the PutBlock have finished.
- *
- * CloseContainer is synchronized with WriteChunk and PutBlock operations,
- * CloseContainer for a container is processed after all the preceding write
- * operations for the container have finished.
- * */
+ **/
+
 public class ContainerStateMachine extends BaseStateMachine {
-  static final Logger LOG = LoggerFactory.getLogger(
-      ContainerStateMachine.class);
-  private final SimpleStateMachineStorage storage
-      = new SimpleStateMachineStorage();
+  static final Logger LOG =
+      LoggerFactory.getLogger(ContainerStateMachine.class);
+  private final SimpleStateMachineStorage storage =
+      new SimpleStateMachineStorage();
   private final ContainerDispatcher dispatcher;
   private ThreadPoolExecutor chunkExecutor;
   private final XceiverServerRatis ratisServer;
   private final ConcurrentHashMap<Long, CompletableFuture<Message>>
       writeChunkFutureMap;
-  private final ConcurrentHashMap<Long, StateMachineHelper> stateMachineMap;
+  private final ConcurrentHashMap<Long, CompletableFuture<Message>>
+      createContainerFutureMap;
+  private ExecutorService[] executors;
+  private final int numExecutors;
   /**
    * CSM metrics.
    */
   private final CSMMetrics metrics;
 
   public ContainerStateMachine(ContainerDispatcher dispatcher,
-      ThreadPoolExecutor chunkExecutor, XceiverServerRatis ratisServer) {
+      ThreadPoolExecutor chunkExecutor, XceiverServerRatis ratisServer,
+      int  numOfExecutors) {
     this.dispatcher = dispatcher;
     this.chunkExecutor = chunkExecutor;
     this.ratisServer = ratisServer;
     this.writeChunkFutureMap = new ConcurrentHashMap<>();
-    this.stateMachineMap = new ConcurrentHashMap<>();
     metrics = CSMMetrics.create();
+    this.createContainerFutureMap = new ConcurrentHashMap<>();
+    this.numExecutors = numOfExecutors;
+    executors = new ExecutorService[numExecutors];
+    for (int i = 0; i < numExecutors; i++) {
+      executors[i] = Executors.newSingleThreadExecutor();
+    }
   }
 
   @Override
@@ -229,6 +227,41 @@ public class ContainerStateMachine extends BaseStateMachine {
     return dispatchCommand(requestProto)::toByteString;
   }
 
+  private ExecutorService getCommandExecutor(
+      ContainerCommandRequestProto requestProto) {
+    int executorId = (int)(requestProto.getContainerID() % numExecutors);
+    return executors[executorId];
+  }
+
+  private CompletableFuture<Message> handleWriteChunk(
+      ContainerCommandRequestProto requestProto, long entryIndex) {
+    final WriteChunkRequestProto write = requestProto.getWriteChunk();
+    long containerID = write.getBlockID().getContainerID();
+    CompletableFuture<Message> future =
+        createContainerFutureMap.get(containerID);
+    CompletableFuture<Message> writeChunkFuture;
+    if (future != null) {
+      writeChunkFuture = future.thenApplyAsync(
+          v -> runCommand(requestProto), chunkExecutor);
+    } else {
+      writeChunkFuture = CompletableFuture.supplyAsync(
+          () -> runCommand(requestProto), chunkExecutor);
+    }
+    writeChunkFutureMap.put(entryIndex, writeChunkFuture);
+    // Remove the future once it finishes execution from the
+    // writeChunkFutureMap.
+    writeChunkFuture.thenApply(r -> writeChunkFutureMap.remove(entryIndex));
+    return writeChunkFuture;
+  }
+
+  private CompletableFuture<Message> handleCreateContainer(
+      ContainerCommandRequestProto requestProto) {
+    long containerID = requestProto.getContainerID();
+    createContainerFutureMap.
+        computeIfAbsent(containerID, k -> new CompletableFuture<>());
+    return CompletableFuture.completedFuture(() -> ByteString.EMPTY);
+  }
+
   /*
    * writeStateMachineData calls are not synchronized with each other
    * and also with applyTransaction.
@@ -240,17 +273,15 @@ public class ContainerStateMachine extends BaseStateMachine {
       final ContainerCommandRequestProto requestProto =
           getRequestProto(entry.getSmLogEntry().getStateMachineData());
       Type cmdType = requestProto.getCmdType();
-      long containerId = requestProto.getContainerID();
-      stateMachineMap
-          .computeIfAbsent(containerId, k -> new StateMachineHelper());
-      CompletableFuture<Message> stateMachineFuture =
-          stateMachineMap.get(containerId)
-              .handleStateMachineData(requestProto, entry.getIndex());
-      if (stateMachineFuture == null) {
-        throw new IllegalStateException(
-            "Cmd Type:" + cmdType + " should not have state machine data");
+      switch (cmdType) {
+      case CreateContainer:
+        return handleCreateContainer(requestProto);
+      case WriteChunk:
+        return handleWriteChunk(requestProto, entry.getIndex());
+      default:
+        throw new IllegalStateException("Cmd Type:" + cmdType
+            + " should not have state machine data");
       }
-      return stateMachineFuture;
     } catch (IOException e) {
       metrics.incNumWriteStateMachineFails();
       return completeExceptionally(e);
@@ -270,14 +301,14 @@ public class ContainerStateMachine extends BaseStateMachine {
     }
   }
 
-  private LogEntryProto readStateMachineData(LogEntryProto entry,
+  private ByteString readStateMachineData(LogEntryProto entry,
       ContainerCommandRequestProto requestProto) {
     WriteChunkRequestProto writeChunkRequestProto =
         requestProto.getWriteChunk();
     // Assert that store log entry is for COMMIT_DATA, the WRITE_DATA is
     // written through writeStateMachineData.
-    Preconditions.checkArgument(writeChunkRequestProto.getStage()
-        == Stage.COMMIT_DATA);
+    Preconditions
+        .checkArgument(writeChunkRequestProto.getStage() == Stage.COMMIT_DATA);
 
     // prepare the chunk to be read
     ReadChunkRequestProto.Builder readChunkRequestProto =
@@ -286,8 +317,7 @@ public class ContainerStateMachine extends BaseStateMachine {
             .setChunkData(writeChunkRequestProto.getChunkData());
     ContainerCommandRequestProto dataContainerCommandProto =
         ContainerCommandRequestProto.newBuilder(requestProto)
-            .setCmdType(Type.ReadChunk)
-            .setReadChunk(readChunkRequestProto)
+            .setCmdType(Type.ReadChunk).setReadChunk(readChunkRequestProto)
             .build();
 
     // read the chunk
@@ -302,25 +332,13 @@ public class ContainerStateMachine extends BaseStateMachine {
     final WriteChunkRequestProto.Builder dataWriteChunkProto =
         WriteChunkRequestProto.newBuilder(writeChunkRequestProto)
             // adding the state machine data
-            .setData(responseProto.getData())
-            .setStage(Stage.WRITE_DATA);
+            .setData(responseProto.getData()).setStage(Stage.WRITE_DATA);
 
     ContainerCommandRequestProto.Builder newStateMachineProto =
         ContainerCommandRequestProto.newBuilder(requestProto)
             .setWriteChunk(dataWriteChunkProto);
 
-    return recreateLogEntryProto(entry,
-        newStateMachineProto.build().toByteString());
-  }
-
-  private LogEntryProto recreateLogEntryProto(LogEntryProto entry,
-      ByteString stateMachineData) {
-    // recreate the log entry
-    final SMLogEntryProto log =
-        SMLogEntryProto.newBuilder(entry.getSmLogEntry())
-            .setStateMachineData(stateMachineData)
-            .build();
-    return LogEntryProto.newBuilder(entry).setSmLogEntry(log).build();
+    return newStateMachineProto.build().toByteString();
   }
 
   /**
@@ -347,11 +365,11 @@ public class ContainerStateMachine extends BaseStateMachine {
    * evicted.
    */
   @Override
-  public CompletableFuture<LogEntryProto> readStateMachineData(
+  public CompletableFuture<ByteString> readStateMachineData(
       LogEntryProto entry) {
     SMLogEntryProto smLogEntryProto = entry.getSmLogEntry();
     if (!smLogEntryProto.getStateMachineData().isEmpty()) {
-      return CompletableFuture.completedFuture(entry);
+      return CompletableFuture.completedFuture(ByteString.EMPTY);
     }
 
     try {
@@ -365,9 +383,7 @@ public class ContainerStateMachine extends BaseStateMachine {
                 readStateMachineData(entry, requestProto),
             chunkExecutor);
       } else if (requestProto.getCmdType() == Type.CreateContainer) {
-        LogEntryProto log =
-            recreateLogEntryProto(entry, requestProto.toByteString());
-        return CompletableFuture.completedFuture(log);
+        return CompletableFuture.completedFuture(requestProto.toByteString());
       } else {
         throw new IllegalStateException("Cmd type:" + requestProto.getCmdType()
             + " cannot have state machine data");
@@ -387,13 +403,44 @@ public class ContainerStateMachine extends BaseStateMachine {
       metrics.incNumApplyTransactionsOps();
       ContainerCommandRequestProto requestProto =
           getRequestProto(trx.getSMLogEntry().getData());
-      Preconditions.checkState(!HddsUtils.isReadOnly(requestProto));
-      stateMachineMap.computeIfAbsent(requestProto.getContainerID(),
-          k -> new StateMachineHelper());
-      long index =
-          trx.getLogEntry() == null ? -1 : trx.getLogEntry().getIndex();
-      return stateMachineMap.get(requestProto.getContainerID())
-          .executeContainerCommand(requestProto, index);
+      Type cmdType = requestProto.getCmdType();
+      CompletableFuture<Message> future;
+      if (cmdType == Type.PutBlock) {
+        BlockData blockData;
+        ContainerProtos.BlockData blockDataProto =
+            requestProto.getPutBlock().getBlockData();
+
+        // set the blockCommitSequenceId
+        try {
+          blockData = BlockData.getFromProtoBuf(blockDataProto);
+        } catch (IOException ioe) {
+          LOG.error("unable to retrieve blockData info for Block {}",
+              blockDataProto.getBlockID());
+          return completeExceptionally(ioe);
+        }
+        blockData.setBlockCommitSequenceId(trx.getLogEntry().getIndex());
+        final ContainerProtos.PutBlockRequestProto putBlockRequestProto =
+            ContainerProtos.PutBlockRequestProto
+                .newBuilder(requestProto.getPutBlock())
+                .setBlockData(blockData.getProtoBufMessage()).build();
+        ContainerCommandRequestProto containerCommandRequestProto =
+            ContainerCommandRequestProto.newBuilder(requestProto)
+                .setPutBlock(putBlockRequestProto).build();
+        future = CompletableFuture
+            .supplyAsync(() -> runCommand(containerCommandRequestProto),
+                getCommandExecutor(requestProto));
+      } else {
+        future = CompletableFuture.supplyAsync(() -> runCommand(requestProto),
+            getCommandExecutor(requestProto));
+      }
+      // Mark the createContainerFuture complete so that writeStateMachineData
+      // for WriteChunk gets unblocked
+      if (cmdType == Type.CreateContainer) {
+        long containerID = requestProto.getContainerID();
+        future.thenApply(
+            r -> createContainerFutureMap.remove(containerID).complete(null));
+      }
+      return future;
     } catch (IOException e) {
       metrics.incNumApplyTransactionsFails();
       return completeExceptionally(e);
@@ -419,259 +466,8 @@ public class ContainerStateMachine extends BaseStateMachine {
 
   @Override
   public void close() throws IOException {
-  }
-
-  /**
-   * Class to manage the future tasks for writeChunks.
-   */
-  static class CommitChunkFutureMap {
-    private final ConcurrentHashMap<Long, CompletableFuture<Message>>
-        block2ChunkMap = new ConcurrentHashMap<>();
-
-    synchronized int removeAndGetSize(long index) {
-      block2ChunkMap.remove(index);
-      return block2ChunkMap.size();
+    for (int i = 0; i < numExecutors; i++){
+      executors[i].shutdown();
     }
-
-    synchronized CompletableFuture<Message> add(long index,
-        CompletableFuture<Message> future) {
-      return block2ChunkMap.put(index, future);
-    }
-
-    synchronized List<CompletableFuture<Message>> getAll() {
-      return new ArrayList<>(block2ChunkMap.values());
-    }
-  }
-
-  /**
-   * This class maintains maps and provide utilities to enforce synchronization
-   * among createContainer, writeChunk, putBlock and closeContainer.
-   */
-  private class StateMachineHelper {
-
-    private CompletableFuture<Message> createContainerFuture;
-
-    // Map for maintaining all writeChunk futures mapped to blockId
-    private final ConcurrentHashMap<Long, CommitChunkFutureMap>
-        block2ChunkMap;
-
-    // Map for putBlock futures
-    private final ConcurrentHashMap<Long, CompletableFuture<Message>>
-        blockCommitMap;
-
-    StateMachineHelper() {
-      createContainerFuture = null;
-      block2ChunkMap = new ConcurrentHashMap<>();
-      blockCommitMap = new ConcurrentHashMap<>();
-    }
-
-    // The following section handles writeStateMachineData transactions
-    // on a container
-
-    // enqueue the create container future during writeStateMachineData
-    // so that the write stateMachine data phase of writeChunk wait on
-    // create container to finish.
-    private CompletableFuture<Message> handleCreateContainer() {
-      createContainerFuture = new CompletableFuture<>();
-      return CompletableFuture.completedFuture(() -> ByteString.EMPTY);
-    }
-
-    // This synchronizes on create container to finish
-    private CompletableFuture<Message> handleWriteChunk(
-        ContainerCommandRequestProto requestProto, long entryIndex) {
-      CompletableFuture<Message> containerOpFuture;
-
-      if (createContainerFuture != null) {
-        containerOpFuture = createContainerFuture
-            .thenApplyAsync(v -> runCommand(requestProto), chunkExecutor);
-      } else {
-        containerOpFuture = CompletableFuture
-            .supplyAsync(() -> runCommand(requestProto), chunkExecutor);
-      }
-      writeChunkFutureMap.put(entryIndex, containerOpFuture);
-      return containerOpFuture;
-    }
-
-    CompletableFuture<Message> handleStateMachineData(
-        final ContainerCommandRequestProto requestProto, long index) {
-      Type cmdType = requestProto.getCmdType();
-      if (cmdType == Type.CreateContainer) {
-        return handleCreateContainer();
-      } else if (cmdType == Type.WriteChunk) {
-        return handleWriteChunk(requestProto, index);
-      } else {
-        return null;
-      }
-    }
-
-    // The following section handles applyTransaction transactions
-    // on a container
-
-    private CompletableFuture<Message> handlePutBlock(
-        ContainerCommandRequestProto requestProto, long index) {
-      List<CompletableFuture<Message>> futureList = new ArrayList<>();
-      BlockData blockData = null;
-      ContainerProtos.BlockData blockDataProto =
-          requestProto.getPutBlock().getBlockData();
-
-      // set the blockCommitSequenceId
-      try {
-        blockData = BlockData.getFromProtoBuf(blockDataProto);
-      } catch (IOException ioe) {
-        LOG.error("unable to retrieve blockData info for Block {}",
-            blockDataProto.getBlockID());
-        return completeExceptionally(ioe);
-      }
-      blockData.setBlockCommitSequenceId(index);
-      final ContainerProtos.PutBlockRequestProto putBlockRequestProto =
-          ContainerProtos.PutBlockRequestProto
-              .newBuilder(requestProto.getPutBlock())
-              .setBlockData(blockData.getProtoBufMessage()).build();
-      ContainerCommandRequestProto containerCommandRequestProto =
-          ContainerCommandRequestProto.newBuilder(requestProto)
-              .setPutBlock(putBlockRequestProto).build();
-      long localId = blockDataProto.getBlockID().getLocalID();
-      // Need not wait for create container future here as it has already
-      // finished.
-      if (block2ChunkMap.get(localId) != null) {
-        futureList.addAll(block2ChunkMap.get(localId).getAll());
-      }
-      CompletableFuture<Message> effectiveFuture =
-          runCommandAfterFutures(futureList, containerCommandRequestProto);
-
-      CompletableFuture<Message> putBlockFuture =
-          effectiveFuture.thenApply(message -> {
-            blockCommitMap.remove(localId);
-            return message;
-          });
-      blockCommitMap.put(localId, putBlockFuture);
-      return putBlockFuture;
-    }
-
-    // Close Container should be executed only if all pending WriteType
-    // container cmds get executed. Transactions which can return a future
-    // are WriteChunk and PutBlock.
-    private CompletableFuture<Message> handleCloseContainer(
-        ContainerCommandRequestProto requestProto) {
-      List<CompletableFuture<Message>> futureList = new ArrayList<>();
-
-      // No need to wait for create container future here as it should have
-      // already finished.
-      block2ChunkMap.values().forEach(b -> futureList.addAll(b.getAll()));
-      futureList.addAll(blockCommitMap.values());
-
-      // There are pending write Chunk/PutBlock type requests
-      // Queue this closeContainer request behind all these requests
-      CompletableFuture<Message> closeContainerFuture =
-          runCommandAfterFutures(futureList, requestProto);
-
-      return closeContainerFuture.thenApply(message -> {
-        stateMachineMap.remove(requestProto.getContainerID());
-        return message;
-      });
-    }
-
-    private CompletableFuture<Message> handleChunkCommit(
-        ContainerCommandRequestProto requestProto, long index) {
-      WriteChunkRequestProto write = requestProto.getWriteChunk();
-      // the data field has already been removed in start Transaction
-      Preconditions.checkArgument(!write.hasData());
-      CompletableFuture<Message> stateMachineFuture =
-          writeChunkFutureMap.remove(index);
-      CompletableFuture<Message> commitChunkFuture = stateMachineFuture
-          .thenComposeAsync(v -> CompletableFuture
-              .completedFuture(runCommand(requestProto)));
-
-      long localId = requestProto.getWriteChunk().getBlockID().getLocalID();
-      // Put the applyTransaction Future again to the Map.
-      // closeContainer should synchronize with this.
-      block2ChunkMap
-          .computeIfAbsent(localId, id -> new CommitChunkFutureMap())
-          .add(index, commitChunkFuture);
-      return commitChunkFuture.thenApply(message -> {
-        block2ChunkMap.computeIfPresent(localId, (containerId, chunks)
-            -> chunks.removeAndGetSize(index) == 0? null: chunks);
-        return message;
-      });
-    }
-
-    private CompletableFuture<Message> runCommandAfterFutures(
-        List<CompletableFuture<Message>> futureList,
-        ContainerCommandRequestProto requestProto) {
-      CompletableFuture<Message> effectiveFuture;
-      if (futureList.isEmpty()) {
-        effectiveFuture = CompletableFuture
-            .supplyAsync(() -> runCommand(requestProto));
-
-      } else {
-        CompletableFuture<Void> allFuture = CompletableFuture.allOf(
-            futureList.toArray(new CompletableFuture[futureList.size()]));
-        effectiveFuture = allFuture
-            .thenApplyAsync(v -> runCommand(requestProto));
-      }
-      return effectiveFuture;
-    }
-
-    CompletableFuture<Message> handleCreateContainer(
-        ContainerCommandRequestProto requestProto) {
-      CompletableFuture<Message> future =
-          CompletableFuture.completedFuture(runCommand(requestProto));
-      future.thenAccept(m -> {
-        createContainerFuture.complete(m);
-        createContainerFuture = null;
-      });
-      return future;
-    }
-
-    CompletableFuture<Message> handleOtherCommands(
-        ContainerCommandRequestProto requestProto) {
-      return CompletableFuture.completedFuture(runCommand(requestProto));
-    }
-
-    CompletableFuture<Message> executeContainerCommand(
-        ContainerCommandRequestProto requestProto, long index) {
-      Type cmdType = requestProto.getCmdType();
-      switch (cmdType) {
-      case WriteChunk:
-        return handleChunkCommit(requestProto, index);
-      case CloseContainer:
-        return handleCloseContainer(requestProto);
-      case PutBlock:
-        return handlePutBlock(requestProto, index);
-      case CreateContainer:
-        return handleCreateContainer(requestProto);
-      default:
-        return handleOtherCommands(requestProto);
-      }
-    }
-  }
-
-  @VisibleForTesting
-  public ConcurrentHashMap<Long, StateMachineHelper> getStateMachineMap() {
-    return stateMachineMap;
-  }
-
-  @VisibleForTesting
-  public CompletableFuture<Message> getCreateContainerFuture(long containerId) {
-    StateMachineHelper helper = stateMachineMap.get(containerId);
-    return helper == null ? null : helper.createContainerFuture;
-  }
-
-  @VisibleForTesting
-  public List<CompletableFuture<Message>> getCommitChunkFutureMap(
-      long containerId) {
-    StateMachineHelper helper = stateMachineMap.get(containerId);
-    if (helper != null) {
-      List<CompletableFuture<Message>> futureList = new ArrayList<>();
-      stateMachineMap.get(containerId).block2ChunkMap.values()
-          .forEach(b -> futureList.addAll(b.getAll()));
-      return futureList;
-    }
-    return null;
-  }
-
-  @VisibleForTesting
-  public Collection<CompletableFuture<Message>> getWriteChunkFutureMap() {
-    return writeChunkFutureMap.values();
   }
 }

http://git-wip-us.apache.org/repos/asf/hadoop/blob/0bc6d048/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/XceiverServerRatis.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/XceiverServerRatis.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/XceiverServerRatis.java
index c2ef504..9094217 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/XceiverServerRatis.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/XceiverServerRatis.java
@@ -56,9 +56,9 @@ import org.apache.ratis.rpc.RpcType;
 import org.apache.ratis.rpc.SupportedRpcType;
 import org.apache.ratis.server.RaftServer;
 import org.apache.ratis.server.RaftServerConfigKeys;
-import org.apache.ratis.shaded.proto.RaftProtos;
-import org.apache.ratis.shaded.proto.RaftProtos.RoleInfoProto;
-import org.apache.ratis.shaded.proto.RaftProtos.ReplicationLevel;
+import org.apache.ratis.proto.RaftProtos;
+import org.apache.ratis.proto.RaftProtos.RoleInfoProto;
+import org.apache.ratis.proto.RaftProtos.ReplicationLevel;
 import org.apache.ratis.util.SizeInBytes;
 import org.apache.ratis.util.TimeDuration;
 import org.slf4j.Logger;
@@ -97,6 +97,7 @@ public final class XceiverServerRatis implements XceiverServerSpi {
   private final StateContext context;
   private final ReplicationLevel replicationLevel;
   private long nodeFailureTimeoutMs;
+  private ContainerStateMachine stateMachine;
 
   private XceiverServerRatis(DatanodeDetails dd, int port,
       ContainerDispatcher dispatcher, Configuration conf, StateContext context)
@@ -112,12 +113,15 @@ public final class XceiverServerRatis implements XceiverServerSpi {
             100, TimeUnit.SECONDS,
             new ArrayBlockingQueue<>(1024),
             new ThreadPoolExecutor.CallerRunsPolicy());
+    final int numContainerOpExecutors = conf.getInt(
+        OzoneConfigKeys.DFS_CONTAINER_RATIS_NUM_CONTAINER_OP_EXECUTORS_KEY,
+        OzoneConfigKeys.DFS_CONTAINER_RATIS_NUM_CONTAINER_OP_EXECUTORS_DEFAULT);
     this.context = context;
     this.replicationLevel =
         conf.getEnum(OzoneConfigKeys.DFS_CONTAINER_RATIS_REPLICATION_LEVEL_KEY,
             OzoneConfigKeys.DFS_CONTAINER_RATIS_REPLICATION_LEVEL_DEFAULT);
-    ContainerStateMachine stateMachine =
-        new ContainerStateMachine(dispatcher, chunkExecutor, this);
+    stateMachine = new ContainerStateMachine(dispatcher, chunkExecutor, this,
+        numContainerOpExecutors);
     this.server = RaftServer.newBuilder()
         .setServerId(RatisHelper.toRaftPeerId(dd))
         .setProperties(serverProperties)
@@ -292,6 +296,7 @@ public final class XceiverServerRatis implements XceiverServerSpi {
   public void stop() {
     try {
       chunkExecutor.shutdown();
+      stateMachine.close();
       server.close();
     } catch (IOException e) {
       throw new RuntimeException(e);

http://git-wip-us.apache.org/repos/asf/hadoop/blob/0bc6d048/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/helpers/ChunkUtils.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/helpers/ChunkUtils.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/helpers/ChunkUtils.java
index 62e328e..8bdae0f 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/helpers/ChunkUtils.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/helpers/ChunkUtils.java
@@ -35,7 +35,7 @@ import org.apache.hadoop.ozone.container.common.helpers.ChunkInfo;
 import org.apache.hadoop.ozone.container.common.helpers.ContainerUtils;
 import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainerData;
 import org.apache.hadoop.ozone.container.keyvalue.impl.ChunkManagerImpl;
-import org.apache.ratis.shaded.com.google.protobuf.ByteString;
+import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
 import org.apache.hadoop.ozone.container.common.volume.VolumeIOStats;
 import org.apache.hadoop.util.Time;
 import org.slf4j.Logger;

http://git-wip-us.apache.org/repos/asf/hadoop/blob/0bc6d048/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/helpers/SmallFileUtils.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/helpers/SmallFileUtils.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/helpers/SmallFileUtils.java
index 3495363..dee0c11 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/helpers/SmallFileUtils.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/helpers/SmallFileUtils.java
@@ -21,7 +21,7 @@ import com.google.common.base.Preconditions;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
 import org.apache.hadoop.ozone.container.common.helpers.ChunkInfo;
 import org.apache.hadoop.ozone.container.common.helpers.ContainerUtils;
-import org.apache.ratis.shaded.com.google.protobuf.ByteString;
+import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos
     .ContainerCommandRequestProto;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos

http://git-wip-us.apache.org/repos/asf/hadoop/blob/0bc6d048/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/statemachine/background/BlockDeletingService.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/statemachine/background/BlockDeletingService.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/statemachine/background/BlockDeletingService.java
index d96fbfa..61a303f 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/statemachine/background/BlockDeletingService.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/statemachine/background/BlockDeletingService.java
@@ -27,7 +27,7 @@ import org.apache.hadoop.ozone.container.common.interfaces.ContainerDeletionChoo
 import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainerData;
 import org.apache.hadoop.ozone.container.keyvalue.helpers.BlockUtils;
 import org.apache.hadoop.util.ReflectionUtils;
-import org.apache.ratis.shaded.com.google.protobuf
+import org.apache.ratis.thirdparty.com.google.protobuf
     .InvalidProtocolBufferException;
 import org.apache.commons.io.FileUtils;
 import org.apache.hadoop.conf.Configuration;

http://git-wip-us.apache.org/repos/asf/hadoop/blob/0bc6d048/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/GrpcReplicationClient.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/GrpcReplicationClient.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/GrpcReplicationClient.java
index 3aafb0c..c8a40b2 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/GrpcReplicationClient.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/GrpcReplicationClient.java
@@ -36,9 +36,9 @@ import org.apache.hadoop.hdds.protocol.datanode.proto
 import org.apache.hadoop.ozone.OzoneConfigKeys;
 
 import com.google.common.base.Preconditions;
-import org.apache.ratis.shaded.io.grpc.ManagedChannel;
-import org.apache.ratis.shaded.io.grpc.netty.NettyChannelBuilder;
-import org.apache.ratis.shaded.io.grpc.stub.StreamObserver;
+import org.apache.ratis.thirdparty.io.grpc.ManagedChannel;
+import org.apache.ratis.thirdparty.io.grpc.netty.NettyChannelBuilder;
+import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

http://git-wip-us.apache.org/repos/asf/hadoop/blob/0bc6d048/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/GrpcReplicationService.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/GrpcReplicationService.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/GrpcReplicationService.java
index d8f696f..30a251d 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/GrpcReplicationService.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/GrpcReplicationService.java
@@ -30,8 +30,8 @@ import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos
 import org.apache.hadoop.hdds.protocol.datanode.proto
     .IntraDatanodeProtocolServiceGrpc;
 
-import org.apache.ratis.shaded.com.google.protobuf.ByteString;
-import org.apache.ratis.shaded.io.grpc.stub.StreamObserver;
+import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
+import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

http://git-wip-us.apache.org/repos/asf/hadoop/blob/0bc6d048/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestHddsDispatcher.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestHddsDispatcher.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestHddsDispatcher.java
index fc622b2..fc84ae7 100644
--- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestHddsDispatcher.java
+++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestHddsDispatcher.java
@@ -40,7 +40,7 @@ import org.apache.hadoop.ozone.container.common.volume.VolumeSet;
 import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainer;
 import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainerData;
 import org.apache.hadoop.test.GenericTestUtils;
-import org.apache.ratis.shaded.com.google.protobuf.ByteString;
+import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
 import org.junit.Assert;
 import org.junit.Test;
 import org.mockito.Mockito;

http://git-wip-us.apache.org/repos/asf/hadoop/blob/0bc6d048/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkGroupOutputStream.java
----------------------------------------------------------------------
diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkGroupOutputStream.java b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkGroupOutputStream.java
index 6580c2c9..de666ce 100644
--- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkGroupOutputStream.java
+++ b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkGroupOutputStream.java
@@ -712,6 +712,11 @@ public class ChunkGroupOutputStream extends OutputStream {
       if (this.outputStream instanceof ChunkOutputStream) {
         ChunkOutputStream out = (ChunkOutputStream) this.outputStream;
         return out.getBlockCommitSequenceId();
+      } else if (outputStream == null) {
+        // For a pre allocated block for which no write has been initiated,
+        // the OutputStream will be null here.
+        // In such cases, the default blockCommitSequenceId will be 0
+        return 0;
       }
       throw new IOException("Invalid Output Stream for Key: " + key);
     }

http://git-wip-us.apache.org/repos/asf/hadoop/blob/0bc6d048/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestCloseContainerHandlingByClient.java
----------------------------------------------------------------------
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestCloseContainerHandlingByClient.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestCloseContainerHandlingByClient.java
index f606263..da8d334 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestCloseContainerHandlingByClient.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestCloseContainerHandlingByClient.java
@@ -341,7 +341,6 @@ public class TestCloseContainerHandlingByClient {
     Assert.assertTrue(key.getOutputStream() instanceof ChunkGroupOutputStream);
     // With the initial size provided, it should have pre allocated 4 blocks
     Assert.assertEquals(2, groupOutputStream.getStreamEntries().size());
-    Assert.assertEquals(2, groupOutputStream.getLocationInfoList().size());
     String dataString = fixedLengthString(keyString, (1 * blockSize));
     byte[] data = dataString.getBytes();
     key.write(data);

http://git-wip-us.apache.org/repos/asf/hadoop/blob/0bc6d048/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/ContainerTestHelper.java
----------------------------------------------------------------------
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/ContainerTestHelper.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/ContainerTestHelper.java
index f278479..324187c 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/ContainerTestHelper.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/ContainerTestHelper.java
@@ -25,7 +25,7 @@ import org.apache.hadoop.ozone.HddsDatanodeService;
 import org.apache.hadoop.ozone.MiniOzoneCluster;
 import org.apache.hadoop.ozone.container.common.impl.ContainerData;
 import org.apache.hadoop.ozone.container.common.interfaces.Container;
-import org.apache.ratis.shaded.com.google.protobuf.ByteString;
+import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
 import org.apache.commons.codec.binary.Hex;
 import org.apache.hadoop.hdds.client.BlockID;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -439,6 +439,7 @@ public final class ContainerTestHelper {
     List<ContainerProtos.ChunkInfo> newList = new LinkedList<>();
     newList.add(writeRequest.getChunkData());
     blockData.setChunks(newList);
+    blockData.setBlockCommitSequenceId(0);
     putRequest.setBlockData(blockData.getProtoBufMessage());
 
     ContainerCommandRequestProto.Builder request =

http://git-wip-us.apache.org/repos/asf/hadoop/blob/0bc6d048/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestCloseContainerHandler.java
----------------------------------------------------------------------
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestCloseContainerHandler.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestCloseContainerHandler.java
index 78bf008..92bad27 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestCloseContainerHandler.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestCloseContainerHandler.java
@@ -32,7 +32,7 @@ import org.apache.hadoop.ozone.container.common.volume.VolumeSet;
 import org.apache.hadoop.ozone.container.keyvalue.KeyValueHandler;
 import org.apache.hadoop.ozone.container.common.helpers.ChunkInfo;
 import org.apache.hadoop.hdds.scm.container.common.helpers.Pipeline;
-import org.apache.ratis.shaded.com.google.protobuf.ByteString;
+import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.Rule;

http://git-wip-us.apache.org/repos/asf/hadoop/blob/0bc6d048/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/server/TestContainerServer.java
----------------------------------------------------------------------
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/server/TestContainerServer.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/server/TestContainerServer.java
index de55d9e..e6ebbf1 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/server/TestContainerServer.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/server/TestContainerServer.java
@@ -24,7 +24,6 @@ import org.apache.hadoop.ozone.container.common.interfaces.Handler;
 import org.apache.hadoop.ozone.container.common.volume.VolumeSet;
 import org.apache.hadoop.ozone.container.replication.GrpcReplicationService;
 import org.apache.hadoop.ozone.container.replication.OnDemandContainerReplicationSource;
-import org.apache.ratis.shaded.io.netty.channel.embedded.EmbeddedChannel;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos

http://git-wip-us.apache.org/repos/asf/hadoop/blob/0bc6d048/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/server/TestContainerStateMachine.java
----------------------------------------------------------------------
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/server/TestContainerStateMachine.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/server/TestContainerStateMachine.java
deleted file mode 100644
index c875a7e..0000000
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/server/TestContainerStateMachine.java
+++ /dev/null
@@ -1,201 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.hadoop.ozone.container.server;
-
-
-import org.apache.hadoop.hdds.client.BlockID;
-import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos
-    .ContainerCommandResponseProto;
-import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos
-    .ContainerCommandRequestProto;
-import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos
-    .ContainerType;
-import org.apache.hadoop.hdds.scm.container.common.helpers.Pipeline;
-import org.apache.hadoop.ozone.container.ContainerTestHelper;
-import org.apache.hadoop.ozone.container.common.interfaces.ContainerDispatcher;
-import org.apache.hadoop.ozone.container.common.interfaces.Handler;
-import org.apache.hadoop.ozone.container.common.transport.server.ratis
-    .ContainerStateMachine;
-import org.apache.ratis.RatisHelper;
-import org.apache.ratis.protocol.ClientId;
-import org.apache.ratis.protocol.Message;
-import org.apache.ratis.protocol.RaftClientRequest;
-import org.apache.ratis.shaded.proto.RaftProtos;
-import org.apache.ratis.statemachine.TransactionContext;
-import org.apache.ratis.util.ProtoUtils;
-import org.junit.Assert;
-import org.junit.Test;
-
-import java.io.IOException;
-import java.util.Collections;
-import java.util.Random;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.atomic.AtomicLong;
-
-/**
- * This class tests ContainerStateMachine.
- */
-public class TestContainerStateMachine {
-
-  private static final AtomicLong CALL_ID_COUNTER = new AtomicLong();
-
-  private static long nextCallId() {
-    return CALL_ID_COUNTER.getAndIncrement() & Long.MAX_VALUE;
-  }
-
-  private ThreadPoolExecutor executor =
-      new ThreadPoolExecutor(4, 4, 100, TimeUnit.SECONDS,
-          new ArrayBlockingQueue<>(1024),
-          new ThreadPoolExecutor.CallerRunsPolicy());
-  private ContainerStateMachine stateMachine =
-      new ContainerStateMachine(new TestContainerDispatcher(), executor, null);
-
-
-  @Test
-  public void testCloseContainerSynchronization() throws Exception {
-    Pipeline pipeline = ContainerTestHelper.createPipeline(3);
-    long containerId = new Random().nextLong();
-
-    //create container request
-    RaftClientRequest createContainer = getRaftClientRequest(
-        ContainerTestHelper.getCreateContainerRequest(containerId, pipeline));
-
-    ContainerCommandRequestProto writeChunkProto = ContainerTestHelper
-        .getWriteChunkRequest(pipeline, new BlockID(containerId, nextCallId()),
-            1024);
-
-    RaftClientRequest writeChunkRequest = getRaftClientRequest(writeChunkProto);
-
-    // add putKey request
-    ContainerCommandRequestProto putKeyProto = ContainerTestHelper
-            .getPutBlockRequest(pipeline, writeChunkProto.getWriteChunk());
-    RaftClientRequest putKeyRequest = getRaftClientRequest(putKeyProto);
-
-    TransactionContext createContainerCtxt =
-        startAndWriteStateMachineData(createContainer);
-    // Start and Write into the StateMachine
-    TransactionContext writeChunkcontext =
-        startAndWriteStateMachineData(writeChunkRequest);
-
-    TransactionContext putKeyContext =
-        stateMachine.startTransaction(putKeyRequest);
-    Assert.assertEquals(1, stateMachine.getStateMachineMap().size());
-    Assert.assertNotNull(stateMachine.getCreateContainerFuture(containerId));
-    Assert.assertEquals(1,
-        stateMachine.getWriteChunkFutureMap().size());
-    Assert.assertTrue(
-        stateMachine.getCommitChunkFutureMap(containerId).isEmpty());
-
-    //Add a closeContainerRequest
-    RaftClientRequest closeRequest = getRaftClientRequest(
-        ContainerTestHelper.getCloseContainer(pipeline, containerId));
-
-    TransactionContext closeCtx = stateMachine.startTransaction(closeRequest);
-
-    // Now apply all the transaction for the CreateContainer Command.
-    // This will unblock writeChunks as well
-
-    stateMachine.applyTransaction(createContainerCtxt);
-    stateMachine.applyTransaction(writeChunkcontext);
-    CompletableFuture<Message> putKeyFuture =
-        stateMachine.applyTransaction(putKeyContext);
-    waitForTransactionCompletion(putKeyFuture);
-    // Make sure the putKey transaction complete
-    Assert.assertTrue(putKeyFuture.isDone());
-
-    // Execute the closeContainer. This should ensure all prior Write Type
-    // container requests finish execution
-
-    CompletableFuture<Message> closeFuture =
-        stateMachine.applyTransaction(closeCtx);
-    waitForTransactionCompletion(closeFuture);
-    // Make sure the closeContainer transaction complete
-    Assert.assertTrue(closeFuture.isDone());
-    Assert.assertNull(stateMachine.getCreateContainerFuture(containerId));
-    Assert.assertNull(stateMachine.getCommitChunkFutureMap(containerId));
-
-  }
-
-  private RaftClientRequest getRaftClientRequest(
-      ContainerCommandRequestProto req) throws IOException {
-    ClientId clientId = ClientId.randomId();
-    return new RaftClientRequest(clientId,
-        RatisHelper.toRaftPeerId(ContainerTestHelper.createDatanodeDetails()),
-        RatisHelper.emptyRaftGroup().getGroupId(), nextCallId(), 0,
-        Message.valueOf(req.toByteString()), RaftClientRequest
-        .writeRequestType(RaftProtos.ReplicationLevel.MAJORITY));
-  }
-
-  private void waitForTransactionCompletion(
-      CompletableFuture<Message> future) throws Exception {
-    ExecutorService executorService = Executors.newSingleThreadExecutor();
-    executorService
-        .invokeAll(Collections.singleton(future::get), 10,
-            TimeUnit.SECONDS); // Timeout of 10 minutes.
-    executorService.shutdown();
-  }
-
-  private TransactionContext startAndWriteStateMachineData(
-      RaftClientRequest request) throws IOException {
-    TransactionContext ctx = stateMachine.startTransaction(request);
-    RaftProtos.LogEntryProto e = ProtoUtils
-        .toLogEntryProto(ctx.getSMLogEntry(), request.getSeqNum(),
-            request.getCallId(), ClientId.randomId(), request.getCallId());
-    ctx.setLogEntry(e);
-    stateMachine.writeStateMachineData(e);
-    return ctx;
-  }
-
-  // ContainerDispatcher for test only purpose.
-  private static class TestContainerDispatcher implements ContainerDispatcher {
-    /**
-     * Dispatches commands to container layer.
-     *
-     * @param msg - Command Request
-     * @return Command Response
-     */
-    @Override
-    public ContainerCommandResponseProto dispatch(
-        ContainerCommandRequestProto msg) {
-      return ContainerTestHelper.getCreateContainerResponse(msg);
-    }
-
-    @Override
-    public void init() {
-    }
-
-    @Override
-    public void shutdown() {
-    }
-
-    @Override
-    public void setScmId(String scmId) {
-    }
-
-    @Override
-    public Handler getHandler(ContainerType containerType) {
-      return null;
-    }
-  }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/hadoop/blob/0bc6d048/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/genesis/BenchMarkDatanodeDispatcher.java
----------------------------------------------------------------------
diff --git a/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/genesis/BenchMarkDatanodeDispatcher.java b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/genesis/BenchMarkDatanodeDispatcher.java
index 8811d91..362e53f 100644
--- a/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/genesis/BenchMarkDatanodeDispatcher.java
+++ b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/genesis/BenchMarkDatanodeDispatcher.java
@@ -23,7 +23,7 @@ import org.apache.hadoop.ozone.container.common.statemachine
     .DatanodeStateMachine.DatanodeStates;
 import org.apache.hadoop.ozone.container.common.statemachine.StateContext;
 import org.apache.hadoop.ozone.container.common.volume.VolumeSet;
-import org.apache.ratis.shaded.com.google.protobuf.ByteString;
+import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
 import org.apache.commons.codec.digest.DigestUtils;
 import org.apache.commons.io.FileUtils;
 import org.apache.commons.lang3.RandomStringUtils;

http://git-wip-us.apache.org/repos/asf/hadoop/blob/0bc6d048/hadoop-project/pom.xml
----------------------------------------------------------------------
diff --git a/hadoop-project/pom.xml b/hadoop-project/pom.xml
index 5f88371..7472a54 100644
--- a/hadoop-project/pom.xml
+++ b/hadoop-project/pom.xml
@@ -103,7 +103,7 @@
     <ldap-api.version>1.0.0-M33</ldap-api.version>
 
     <!-- Apache Ratis version -->
-    <ratis.version>0.3.0-eca3531-SNAPSHOT</ratis.version>
+    <ratis.version>0.3.0-9b84d79-SNAPSHOT</ratis.version>
     <jcache.version>1.0-alpha-1</jcache.version>
     <ehcache.version>3.3.1</ehcache.version>
     <hikari.version>2.4.12</hikari.version>


---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org


Mime
View raw message