diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestFailureHandlingByClient.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestFailureHandlingByClient.java
index 4bd4506b55..d95807689b 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestFailureHandlingByClient.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestFailureHandlingByClient.java
@@ -31,6 +31,7 @@
 import org.apache.hadoop.ozone.client.ObjectStore;
 import org.apache.hadoop.ozone.client.OzoneClient;
 import org.apache.hadoop.ozone.client.OzoneClientFactory;
+import org.apache.hadoop.ozone.client.io.BlockOutputStreamEntry;
 import org.apache.hadoop.ozone.client.io.KeyOutputStream;
 import org.apache.hadoop.ozone.client.io.OzoneOutputStream;
 import org.apache.hadoop.ozone.container.ContainerTestHelper;
@@ -38,7 +39,6 @@
 import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
 import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
 import org.junit.Assert;
-import org.junit.Ignore;
 import org.junit.Test;
 
 import java.io.IOException;
@@ -52,9 +52,7 @@
 
 /**
  * Tests Close Container Exception handling by Ozone Client.
- * XXX Disabled [HDDS-1323]
  */
-@Ignore
 public class TestFailureHandlingByClient {
 
   private MiniOzoneCluster cluster;
@@ -214,12 +212,17 @@ public void testMultiBlockWritesWithIntermittentDnFailures()
 
     // get the name of a valid container
     Assert.assertTrue(key.getOutputStream() instanceof KeyOutputStream);
-    KeyOutputStream groupOutputStream =
+    KeyOutputStream keyOutputStream =
         (KeyOutputStream) key.getOutputStream();
-    List<OmKeyLocationInfo> locationInfoList =
-        groupOutputStream.getLocationInfoList();
-    Assert.assertTrue(locationInfoList.size() == 6);
-    long containerId = locationInfoList.get(1).getContainerID();
+    List<BlockOutputStreamEntry> streamEntryList =
+        keyOutputStream.getStreamEntries();
+
+    // Assert that 6 block will be preallocated
+    Assert.assertEquals(6, streamEntryList.size());
+    key.write(data.getBytes());
+    key.flush();
+    long containerId = streamEntryList.get(0).getBlockID().getContainerID();
+    BlockID blockId = streamEntryList.get(0).getBlockID();
     ContainerInfo container =
         cluster.getStorageContainerManager().getContainerManager()
             .getContainer(ContainerID.valueof(containerId));
@@ -243,8 +246,9 @@ public void testMultiBlockWritesWithIntermittentDnFailures()
         .setRefreshPipeline(true)
         .build();
     OmKeyInfo keyInfo = cluster.getOzoneManager().lookupKey(keyArgs);
-    Assert.assertEquals(3 * data.getBytes().length, keyInfo.getDataSize());
-    validateData(keyName, data.concat(data).concat(data).getBytes());
+    Assert.assertEquals(4 * data.getBytes().length, keyInfo.getDataSize());
+    validateData(keyName,
+        data.concat(data).concat(data).concat(data).getBytes());
     shutdown();
   }
 
@@ -309,15 +313,15 @@ public void testContainerExclusionWithClosedContainerException()
     Assert.assertTrue(key.getOutputStream() instanceof KeyOutputStream);
     KeyOutputStream keyOutputStream =
         (KeyOutputStream) key.getOutputStream();
-    List<OmKeyLocationInfo> locationInfoList =
-        keyOutputStream.getLocationInfoList();
+    List<BlockOutputStreamEntry> streamEntryList =
+        keyOutputStream.getStreamEntries();
 
     // Assert that 1 block will be preallocated
-    Assert.assertEquals(1, locationInfoList.size());
+    Assert.assertEquals(1, streamEntryList.size());
     key.write(data.getBytes());
     key.flush();
-    long containerId = locationInfoList.get(0).getContainerID();
-    BlockID blockId = locationInfoList.get(0).getBlockID();
+    long containerId = streamEntryList.get(0).getBlockID().getContainerID();
+    BlockID blockId = streamEntryList.get(0).getBlockID();
     List<Long> containerIdList = new ArrayList<>();
     containerIdList.add(containerId);
 
@@ -368,15 +372,15 @@ public void testDatanodeExclusionWithMajorityCommit() throws Exception {
     Assert.assertTrue(key.getOutputStream() instanceof KeyOutputStream);
     KeyOutputStream keyOutputStream =
         (KeyOutputStream) key.getOutputStream();
-    List<OmKeyLocationInfo> locationInfoList =
-        keyOutputStream.getLocationInfoList();
+    List<BlockOutputStreamEntry> streamEntryList =
+        keyOutputStream.getStreamEntries();
 
     // Assert that 1 block will be preallocated
-    Assert.assertEquals(1, locationInfoList.size());
+    Assert.assertEquals(1, streamEntryList.size());
     key.write(data.getBytes());
     key.flush();
-    long containerId = locationInfoList.get(0).getContainerID();
-    BlockID blockId = locationInfoList.get(0).getBlockID();
+    long containerId = streamEntryList.get(0).getBlockID().getContainerID();
+    BlockID blockId = streamEntryList.get(0).getBlockID();
     ContainerInfo container =
         cluster.getStorageContainerManager().getContainerManager()
             .getContainer(ContainerID.valueof(containerId));
@@ -391,14 +395,16 @@ public void testDatanodeExclusionWithMajorityCommit() throws Exception {
 
     key.write(data.getBytes());
     key.write(data.getBytes());
-    // The close will just write to the buffer
-    key.close();
+    key.flush();
+
     Assert.assertTrue(keyOutputStream.getExcludeList().getDatanodes()
         .contains(datanodes.get(0)));
     Assert.assertTrue(
         keyOutputStream.getExcludeList().getContainerIds().isEmpty());
     Assert.assertTrue(
         keyOutputStream.getExcludeList().getPipelineIds().isEmpty());
+    // The close will just write to the buffer
+    key.close();
 
     OmKeyArgs keyArgs = new OmKeyArgs.Builder().setVolumeName(volumeName)
         .setBucketName(bucketName).setType(HddsProtos.ReplicationType.RATIS)
@@ -430,15 +436,15 @@ public void testPipelineExclusionWithPipelineFailure() throws Exception {
     Assert.assertTrue(key.getOutputStream() instanceof KeyOutputStream);
     KeyOutputStream keyOutputStream =
         (KeyOutputStream) key.getOutputStream();
-    List<OmKeyLocationInfo> locationInfoList =
-        keyOutputStream.getLocationInfoList();
+    List<BlockOutputStreamEntry> streamEntryList =
+        keyOutputStream.getStreamEntries();
 
     // Assert that 1 block will be preallocated
-    Assert.assertEquals(1, locationInfoList.size());
+    Assert.assertEquals(1, streamEntryList.size());
     key.write(data.getBytes());
     key.flush();
-    long containerId = locationInfoList.get(0).getContainerID();
-    BlockID blockId = locationInfoList.get(0).getBlockID();
+    long containerId = streamEntryList.get(0).getBlockID().getContainerID();
+    BlockID blockId = streamEntryList.get(0).getBlockID();
     ContainerInfo container =
         cluster.getStorageContainerManager().getContainerManager()
             .getContainer(ContainerID.valueof(containerId));
@@ -447,21 +453,22 @@ public void testPipelineExclusionWithPipelineFailure() throws Exception {
             .getPipeline(container.getPipelineID());
     List<DatanodeDetails> datanodes = pipeline.getNodes();
 
-    // Two nodes, next write will hit AlraedyClosedException , the pipeline
+    // Two nodes, next write will hit AlreadyClosedException , the pipeline
     // will be added in the exclude list
     cluster.shutdownHddsDatanode(datanodes.get(0));
     cluster.shutdownHddsDatanode(datanodes.get(1));
 
     key.write(data.getBytes());
     key.write(data.getBytes());
-    // The close will just write to the buffer
-    key.close();
+    key.flush();
     Assert.assertTrue(keyOutputStream.getExcludeList().getPipelineIds()
         .contains(pipeline.getId()));
     Assert.assertTrue(
         keyOutputStream.getExcludeList().getContainerIds().isEmpty());
     Assert.assertTrue(
         keyOutputStream.getExcludeList().getDatanodes().isEmpty());
+    // The close will just write to the buffer
+    key.close();
 
     OmKeyArgs keyArgs = new OmKeyArgs.Builder().setVolumeName(volumeName)
         .setBucketName(bucketName).setType(HddsProtos.ReplicationType.RATIS)