diff --git a/hadoop-hdds/client/pom.xml b/hadoop-hdds/client/pom.xml
index 073bd9d..9f2116c 100644
--- a/hadoop-hdds/client/pom.xml
+++ b/hadoop-hdds/client/pom.xml
@@ -46,6 +46,10 @@
       <scope>test</scope>
     </dependency>
     <dependency>
+      <groupId>org.apache.ozone</groupId>
+      <artifactId>hdds-erasurecode</artifactId>
+    </dependency>
+    <dependency>
       <groupId>org.mockito</groupId>
       <artifactId>mockito-core</artifactId>
       <scope>test</scope>
diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/BadDataLocationException.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/BadDataLocationException.java
similarity index 100%
rename from hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/BadDataLocationException.java
rename to hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/BadDataLocationException.java
diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockInputStreamFactory.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockInputStreamFactory.java
similarity index 95%
rename from hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockInputStreamFactory.java
rename to hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockInputStreamFactory.java
index d1bf7f3..6703216 100644
--- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockInputStreamFactory.java
+++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockInputStreamFactory.java
@@ -22,8 +22,8 @@
 import org.apache.hadoop.hdds.scm.XceiverClientFactory;
 import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
 import org.apache.hadoop.hdds.scm.storage.BlockExtendedInputStream;
+import org.apache.hadoop.hdds.scm.storage.BlockLocationInfo;
 import org.apache.hadoop.hdds.security.token.OzoneBlockTokenIdentifier;
-import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
 import org.apache.hadoop.security.token.Token;
 
 import java.util.function.Function;
@@ -47,7 +47,7 @@
    * @return BlockExtendedInputStream of the correct type.
    */
   BlockExtendedInputStream create(ReplicationConfig repConfig,
-      OmKeyLocationInfo blockInfo, Pipeline pipeline,
+      BlockLocationInfo blockInfo, Pipeline pipeline,
       Token<OzoneBlockTokenIdentifier> token, boolean verifyChecksum,
        XceiverClientFactory xceiverFactory,
        Function<BlockID, Pipeline> refreshFunction);
diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockInputStreamFactoryImpl.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockInputStreamFactoryImpl.java
similarity index 96%
rename from hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockInputStreamFactoryImpl.java
rename to hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockInputStreamFactoryImpl.java
index 104e5bc..ba05ec2 100644
--- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockInputStreamFactoryImpl.java
+++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockInputStreamFactoryImpl.java
@@ -25,10 +25,10 @@
 import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
 import org.apache.hadoop.hdds.scm.storage.BlockExtendedInputStream;
 import org.apache.hadoop.hdds.scm.storage.BlockInputStream;
+import org.apache.hadoop.hdds.scm.storage.BlockLocationInfo;
 import org.apache.hadoop.hdds.security.token.OzoneBlockTokenIdentifier;
 import org.apache.hadoop.io.ByteBufferPool;
 import org.apache.hadoop.io.ElasticByteBufferPool;
-import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
 import org.apache.hadoop.security.token.Token;
 
 import java.util.concurrent.ExecutorService;
@@ -75,7 +75,7 @@
    * @return BlockExtendedInputStream of the correct type.
    */
   public BlockExtendedInputStream create(ReplicationConfig repConfig,
-      OmKeyLocationInfo blockInfo, Pipeline pipeline,
+      BlockLocationInfo blockInfo, Pipeline pipeline,
       Token<OzoneBlockTokenIdentifier> token, boolean verifyChecksum,
       XceiverClientFactory xceiverFactory,
       Function<BlockID, Pipeline> refreshFunction) {
diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStream.java
similarity index 97%
rename from hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStream.java
rename to hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStream.java
index b0e9755..70bd384 100644
--- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStream.java
+++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStream.java
@@ -28,8 +28,8 @@
 import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
 import org.apache.hadoop.hdds.scm.pipeline.PipelineID;
 import org.apache.hadoop.hdds.scm.storage.BlockExtendedInputStream;
+import org.apache.hadoop.hdds.scm.storage.BlockLocationInfo;
 import org.apache.hadoop.hdds.scm.storage.ByteReaderStrategy;
-import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -53,7 +53,7 @@
   private final boolean verifyChecksum;
   private final XceiverClientFactory xceiverClientFactory;
   private final Function<BlockID, Pipeline> refreshFunction;
-  private final OmKeyLocationInfo blockInfo;
+  private final BlockLocationInfo blockInfo;
   private final DatanodeDetails[] dataLocations;
   private final BlockExtendedInputStream[] blockStreams;
   private final int maxLocations;
@@ -62,10 +62,6 @@
   private boolean closed = false;
   private boolean seeked = false;
 
-  protected OmKeyLocationInfo getBlockInfo() {
-    return blockInfo;
-  }
-
   protected ECReplicationConfig getRepConfig() {
     return repConfig;
   }
@@ -109,7 +105,7 @@
   }
 
   public ECBlockInputStream(ECReplicationConfig repConfig,
-      OmKeyLocationInfo blockInfo, boolean verifyChecksum,
+      BlockLocationInfo blockInfo, boolean verifyChecksum,
       XceiverClientFactory xceiverClientFactory, Function<BlockID,
       Pipeline> refreshFunction, BlockInputStreamFactory streamFactory) {
     this.repConfig = repConfig;
@@ -174,7 +170,7 @@
           .setState(Pipeline.PipelineState.CLOSED)
           .build();
 
-      OmKeyLocationInfo blkInfo = new OmKeyLocationInfo.Builder()
+      BlockLocationInfo blkInfo = new BlockLocationInfo.Builder()
           .setBlockID(blockInfo.getBlockID())
           .setLength(internalBlockLength(locationIndex + 1))
           .setPipeline(blockInfo.getPipeline())
diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStreamFactory.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStreamFactory.java
similarity index 95%
rename from hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStreamFactory.java
rename to hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStreamFactory.java
index 6c39e93..c9d2b76 100644
--- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStreamFactory.java
+++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStreamFactory.java
@@ -23,7 +23,7 @@
 import org.apache.hadoop.hdds.scm.XceiverClientFactory;
 import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
 import org.apache.hadoop.hdds.scm.storage.BlockExtendedInputStream;
-import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
+import org.apache.hadoop.hdds.scm.storage.BlockLocationInfo;
 
 import java.util.List;
 import java.util.function.Function;
@@ -52,7 +52,7 @@
    */
   BlockExtendedInputStream create(boolean missingLocations,
       List<DatanodeDetails> failedLocations, ReplicationConfig repConfig,
-      OmKeyLocationInfo blockInfo, boolean verifyChecksum,
+      BlockLocationInfo blockInfo, boolean verifyChecksum,
       XceiverClientFactory xceiverFactory,
       Function<BlockID, Pipeline> refreshFunction);
 }
diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStreamFactoryImpl.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStreamFactoryImpl.java
similarity index 97%
rename from hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStreamFactoryImpl.java
rename to hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStreamFactoryImpl.java
index 470df0c..efc3b31 100644
--- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStreamFactoryImpl.java
+++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStreamFactoryImpl.java
@@ -24,8 +24,8 @@
 import org.apache.hadoop.hdds.scm.XceiverClientFactory;
 import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
 import org.apache.hadoop.hdds.scm.storage.BlockExtendedInputStream;
+import org.apache.hadoop.hdds.scm.storage.BlockLocationInfo;
 import org.apache.hadoop.io.ByteBufferPool;
-import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
 
 import java.util.List;
 import java.util.concurrent.ExecutorService;
@@ -75,7 +75,7 @@
    */
   public BlockExtendedInputStream create(boolean missingLocations,
       List<DatanodeDetails> failedLocations, ReplicationConfig repConfig,
-      OmKeyLocationInfo blockInfo, boolean verifyChecksum,
+      BlockLocationInfo blockInfo, boolean verifyChecksum,
       XceiverClientFactory xceiverFactory,
       Function<BlockID, Pipeline> refreshFunction) {
     if (missingLocations) {
diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStreamProxy.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStreamProxy.java
similarity index 97%
rename from hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStreamProxy.java
rename to hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStreamProxy.java
index ecde9c6..49ee7c7 100644
--- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStreamProxy.java
+++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStreamProxy.java
@@ -23,8 +23,8 @@
 import org.apache.hadoop.hdds.scm.XceiverClientFactory;
 import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
 import org.apache.hadoop.hdds.scm.storage.BlockExtendedInputStream;
+import org.apache.hadoop.hdds.scm.storage.BlockLocationInfo;
 import org.apache.hadoop.hdds.scm.storage.ByteReaderStrategy;
-import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -51,7 +51,7 @@
   private final boolean verifyChecksum;
   private final XceiverClientFactory xceiverClientFactory;
   private final Function<BlockID, Pipeline> refreshFunction;
-  private final OmKeyLocationInfo blockInfo;
+  private final BlockLocationInfo blockInfo;
   private final ECBlockInputStreamFactory ecBlockInputStreamFactory;
 
   private BlockExtendedInputStream blockReader;
@@ -96,7 +96,7 @@
   }
 
   public ECBlockInputStreamProxy(ECReplicationConfig repConfig,
-      OmKeyLocationInfo blockInfo, boolean verifyChecksum,
+      BlockLocationInfo blockInfo, boolean verifyChecksum,
       XceiverClientFactory xceiverClientFactory, Function<BlockID,
       Pipeline> refreshFunction, ECBlockInputStreamFactory streamFactory) {
     this.repConfig = repConfig;
diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockReconstructedInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockReconstructedInputStream.java
similarity index 100%
rename from hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockReconstructedInputStream.java
rename to hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockReconstructedInputStream.java
diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockReconstructedStripeInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockReconstructedStripeInputStream.java
similarity index 99%
rename from hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockReconstructedStripeInputStream.java
rename to hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockReconstructedStripeInputStream.java
index d5ec6db..dc7daf8 100644
--- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockReconstructedStripeInputStream.java
+++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockReconstructedStripeInputStream.java
@@ -25,9 +25,9 @@
 import org.apache.hadoop.hdds.scm.XceiverClientFactory;
 import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
 import org.apache.hadoop.hdds.scm.storage.BlockExtendedInputStream;
+import org.apache.hadoop.hdds.scm.storage.BlockLocationInfo;
 import org.apache.hadoop.hdds.scm.storage.ByteReaderStrategy;
 import org.apache.hadoop.io.ByteBufferPool;
-import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
 import org.apache.ozone.erasurecode.rawcoder.RawErasureDecoder;
 import org.apache.ozone.erasurecode.rawcoder.util.CodecUtil;
 import org.apache.ratis.util.Preconditions;
@@ -116,7 +116,7 @@
 
   @SuppressWarnings("checkstyle:ParameterNumber")
   public ECBlockReconstructedStripeInputStream(ECReplicationConfig repConfig,
-      OmKeyLocationInfo blockInfo, boolean verifyChecksum,
+      BlockLocationInfo blockInfo, boolean verifyChecksum,
       XceiverClientFactory xceiverClientFactory, Function<BlockID,
       Pipeline> refreshFunction, BlockInputStreamFactory streamFactory,
       ByteBufferPool byteBufferPool,
diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/InsufficientLocationsException.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/InsufficientLocationsException.java
similarity index 100%
rename from hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/InsufficientLocationsException.java
rename to hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/InsufficientLocationsException.java
diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/package-info.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/package-info.java
new file mode 100644
index 0000000..493ece8
--- /dev/null
+++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/package-info.java
@@ -0,0 +1,23 @@
+/**
+ * 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.client.io;
+
+/**
+ * This package contains Ozone I/O classes.
+ */
diff --git a/hadoop-hdds/common/pom.xml b/hadoop-hdds/common/pom.xml
index 58c0788..cf65d8b 100644
--- a/hadoop-hdds/common/pom.xml
+++ b/hadoop-hdds/common/pom.xml
@@ -71,6 +71,10 @@
       <artifactId>jackson-annotations</artifactId>
     </dependency>
     <dependency>
+      <groupId>com.fasterxml.jackson.datatype</groupId>
+      <artifactId>jackson-datatype-jsr310</artifactId>
+    </dependency>
+    <dependency>
       <groupId>org.apache.ozone</groupId>
       <artifactId>hdds-config</artifactId>
     </dependency>
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 2a962d7..fc83469 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
@@ -403,6 +403,12 @@
   public static final String OZONE_SCM_PIPELINE_CREATION_INTERVAL_DEFAULT =
       "120s";
 
+  public static final String OZONE_SCM_PIPELINE_SCRUB_INTERVAL =
+      "ozone.scm.pipeline.scrub.interval";
+  public static final String OZONE_SCM_PIPELINE_SCRUB_INTERVAL_DEFAULT =
+      "5m";
+
+
   // Allow SCM to auto create factor ONE ratis pipeline.
   public static final String OZONE_SCM_PIPELINE_AUTO_CREATE_FACTOR_ONE =
       "ozone.scm.pipeline.creation.auto.factor.one";
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/ReplicationManagerReport.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/ReplicationManagerReport.java
index 2f2a7bf..60e61b2 100644
--- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/ReplicationManagerReport.java
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/ReplicationManagerReport.java
@@ -107,7 +107,7 @@
     for (HddsProtos.KeyContainerIDList sample : proto.getStatSampleList()) {
       report.setSample(sample.getKey(), sample.getContainerList()
           .stream()
-          .map(c -> ContainerID.getFromProtobuf(c))
+          .map(ContainerID::getFromProtobuf)
           .collect(Collectors.toList()));
     }
     return report;
@@ -147,6 +147,34 @@
   }
 
   /**
+   * Return a map of all stats and their value as a long.
+   * @return
+   */
+  public Map<String, Long> getStats() {
+    Map<String, Long> result = new HashMap<>();
+    for (Map.Entry<String, LongAdder> e : stats.entrySet()) {
+      result.put(e.getKey(), e.getValue().longValue());
+    }
+    return result;
+  }
+
+  /**
+   * Return a map of all samples, with the stat as the key and the samples
+   * for the stat as a List of Long.
+   * @return
+   */
+  public Map<String, List<Long>> getSamples() {
+    Map<String, List<Long>> result = new HashMap<>();
+    for (Map.Entry<String, List<ContainerID>> e : containerSample.entrySet()) {
+      result.put(e.getKey(),
+          e.getValue().stream()
+              .map(c -> c.getId())
+              .collect(Collectors.toList()));
+    }
+    return result;
+  }
+
+  /**
    * Get the stat for the given LifeCycleState. If there is no stat available
    * for that stat -1 is returned.
    * @param stat The requested stat.
@@ -184,7 +212,11 @@
   }
 
   protected void setStat(String stat, long value) {
-    LongAdder adder = getStatAndEnsurePresent(stat);
+    LongAdder adder = stats.get(stat);
+    if (adder == null) {
+      // this is an unknown stat, so ignore it.
+      return;
+    }
     if (adder.longValue() != 0) {
       throw new IllegalStateException(stat + " is expected to be zero");
     }
@@ -192,9 +224,11 @@
   }
 
   protected void setSample(String stat, List<ContainerID> sample) {
-    // First get the stat, as we should not receive a sample for a stat which
-    // does not exist.
-    getStatAndEnsurePresent(stat);
+    LongAdder adder = stats.get(stat);
+    if (adder == null) {
+      // this is an unknown stat, so ignore it.
+      return;
+    }
     // Now check there is not already a sample for this stat
     List<ContainerID> existingSample = containerSample.get(stat);
     if (existingSample != null) {
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockLocationInfo.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockLocationInfo.java
new file mode 100644
index 0000000..286762d
--- /dev/null
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockLocationInfo.java
@@ -0,0 +1,201 @@
+/*
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.hdds.scm.storage;
+
+import org.apache.hadoop.hdds.client.BlockID;
+import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
+import org.apache.hadoop.hdds.security.token.OzoneBlockTokenIdentifier;
+import org.apache.hadoop.security.token.Token;
+
+import java.util.Objects;
+
+/**
+ * One key can be too huge to fit in one container. In which case it gets split
+ * into a number of subkeys. This class represents one such subkey instance.
+ */
+public class BlockLocationInfo {
+  private final BlockID blockID;
+  private long length;
+  private final long offset;
+  // Block token, required for client authentication when security is enabled.
+  private Token<OzoneBlockTokenIdentifier> token;
+  // the version number indicating when this block was added
+  private long createVersion;
+
+  private Pipeline pipeline;
+
+  // PartNumber is set for Multipart upload Keys.
+  private int partNumber;
+
+  protected BlockLocationInfo(Builder builder) {
+    this.blockID = builder.blockID;
+    this.pipeline = builder.pipeline;
+    this.length = builder.length;
+    this.offset = builder.offset;
+    this.token = builder.token;
+    this.partNumber = builder.partNumber;
+    this.createVersion = builder.createVersion;
+  }
+
+  public void setCreateVersion(long version) {
+    createVersion = version;
+  }
+
+  public long getCreateVersion() {
+    return createVersion;
+  }
+
+  public BlockID getBlockID() {
+    return blockID;
+  }
+
+  public long getContainerID() {
+    return blockID.getContainerID();
+  }
+
+  public long getLocalID() {
+    return blockID.getLocalID();
+  }
+
+  public Pipeline getPipeline() {
+    return pipeline;
+  }
+
+  public long getLength() {
+    return length;
+  }
+
+  public void setLength(long length) {
+    this.length = length;
+  }
+
+  public long getOffset() {
+    return offset;
+  }
+
+  public long getBlockCommitSequenceId() {
+    return blockID.getBlockCommitSequenceId();
+  }
+
+  public Token<OzoneBlockTokenIdentifier> getToken() {
+    return token;
+  }
+
+  public void setToken(Token<OzoneBlockTokenIdentifier> token) {
+    this.token = token;
+  }
+
+  public void setPipeline(Pipeline pipeline) {
+    this.pipeline = pipeline;
+  }
+
+  public void setPartNumber(int partNumber) {
+    this.partNumber = partNumber;
+  }
+
+  public int getPartNumber() {
+    return partNumber;
+  }
+
+  /**
+   * Builder of BlockLocationInfo.
+   */
+  public static class Builder {
+    private BlockID blockID;
+    private long length;
+    private long offset;
+    private Token<OzoneBlockTokenIdentifier> token;
+    private Pipeline pipeline;
+    private int partNumber;
+    private long createVersion;
+
+    public Builder setBlockID(BlockID blockId) {
+      this.blockID = blockId;
+      return this;
+    }
+
+    public Builder setPipeline(Pipeline pipeline) {
+      this.pipeline = pipeline;
+      return this;
+    }
+
+    public Builder setLength(long len) {
+      this.length = len;
+      return this;
+    }
+
+    public Builder setOffset(long off) {
+      this.offset = off;
+      return this;
+    }
+
+    public Builder setToken(Token<OzoneBlockTokenIdentifier> bToken) {
+      this.token = bToken;
+      return this;
+    }
+
+    public Builder setPartNumber(int partNum) {
+      this.partNumber = partNum;
+      return this;
+    }
+
+    public Builder setCreateVersion(long version) {
+      this.createVersion = version;
+      return this;
+    }
+
+    public BlockLocationInfo build() {
+      return new BlockLocationInfo(this);
+    }
+  }
+
+  @Override
+  public String  toString() {
+    return "{blockID={containerID=" + blockID.getContainerID() +
+        ", localID=" + blockID.getLocalID() + "}" +
+        ", length=" + length +
+        ", offset=" + offset +
+        ", token=" + token +
+        ", pipeline=" + pipeline +
+        ", createVersion=" + createVersion +
+        ", partNumber=" + partNumber
+        + '}';
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    BlockLocationInfo that = (BlockLocationInfo) o;
+    return length == that.length &&
+        offset == that.offset &&
+        createVersion == that.createVersion &&
+        Objects.equals(blockID, that.blockID) &&
+        Objects.equals(token, that.token) &&
+        Objects.equals(pipeline, that.pipeline);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(blockID, length, offset, token, createVersion,
+        pipeline);
+  }
+}
diff --git a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/server/JsonUtils.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/server/JsonUtils.java
similarity index 100%
rename from hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/server/JsonUtils.java
rename to hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/server/JsonUtils.java
diff --git a/hadoop-hdds/common/src/main/resources/ozone-default.xml b/hadoop-hdds/common/src/main/resources/ozone-default.xml
index 6c76ca6..709cf45 100644
--- a/hadoop-hdds/common/src/main/resources/ozone-default.xml
+++ b/hadoop-hdds/common/src/main/resources/ozone-default.xml
@@ -1369,6 +1369,15 @@
     </description>
   </property>
   <property>
+    <name>ozone.scm.pipeline.scrub.interval</name>
+    <value>5m</value>
+    <tag>OZONE, SCM, PIPELINE</tag>
+    <description>
+      SCM schedules a fixed interval job using the configured interval to
+      scrub pipelines.
+    </description>
+  </property>
+  <property>
     <name>ozone.scm.pipeline.creation.auto.factor.one</name>
     <value>true</value>
     <tag>OZONE, SCM, PIPELINE</tag>
diff --git a/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/scm/container/TestReplicationManagerReport.java b/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/scm/container/TestReplicationManagerReport.java
index a05f9ab..a68f70c 100644
--- a/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/scm/container/TestReplicationManagerReport.java
+++ b/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/scm/container/TestReplicationManagerReport.java
@@ -17,16 +17,22 @@
  */
 package org.apache.hadoop.hdds.scm.container;
 
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.server.JsonUtils;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Random;
 import java.util.concurrent.ThreadLocalRandom;
 
+import static com.fasterxml.jackson.databind.node.JsonNodeType.ARRAY;
+
 /**
  * Tests for the ReplicationManagerReport class.
  */
@@ -64,6 +70,54 @@
         report.getStat(HddsProtos.LifeCycleState.QUASI_CLOSED));
   }
 
+
+  @Test
+  public void testJsonOutput() throws IOException {
+    report.increment(HddsProtos.LifeCycleState.OPEN);
+    report.increment(HddsProtos.LifeCycleState.CLOSED);
+    report.increment(HddsProtos.LifeCycleState.CLOSED);
+
+    report.incrementAndSample(
+        ReplicationManagerReport.HealthState.UNDER_REPLICATED,
+        new ContainerID(1));
+    report.incrementAndSample(
+        ReplicationManagerReport.HealthState.UNDER_REPLICATED,
+        new ContainerID(2));
+    report.incrementAndSample(
+        ReplicationManagerReport.HealthState.OVER_REPLICATED,
+        new ContainerID(3));
+    report.setComplete();
+
+    String jsonString = JsonUtils.toJsonStringWithDefaultPrettyPrinter(report);
+
+    ObjectMapper mapper = new ObjectMapper();
+    JsonNode json = mapper.readTree(jsonString);
+
+    Assert.assertTrue(json.get("reportTimeStamp").longValue() > 0);
+    JsonNode stats = json.get("stats");
+    Assert.assertEquals(1, stats.get("OPEN").longValue());
+    Assert.assertEquals(0, stats.get("CLOSING").longValue());
+    Assert.assertEquals(0, stats.get("QUASI_CLOSED").longValue());
+    Assert.assertEquals(2, stats.get("CLOSED").longValue());
+    Assert.assertEquals(0, stats.get("DELETING").longValue());
+    Assert.assertEquals(0, stats.get("DELETED").longValue());
+
+    Assert.assertEquals(2, stats.get("UNDER_REPLICATED").longValue());
+    Assert.assertEquals(1, stats.get("OVER_REPLICATED").longValue());
+    Assert.assertEquals(0, stats.get("MIS_REPLICATED").longValue());
+    Assert.assertEquals(0, stats.get("MISSING").longValue());
+    Assert.assertEquals(0, stats.get("UNHEALTHY").longValue());
+    Assert.assertEquals(0, stats.get("EMPTY").longValue());
+    Assert.assertEquals(0, stats.get("OPEN_UNHEALTHY").longValue());
+    Assert.assertEquals(0, stats.get("QUASI_CLOSED_STUCK").longValue());
+
+    JsonNode samples = json.get("samples");
+    Assert.assertEquals(ARRAY, samples.get("UNDER_REPLICATED").getNodeType());
+    Assert.assertEquals(1, samples.get("UNDER_REPLICATED").get(0).longValue());
+    Assert.assertEquals(2, samples.get("UNDER_REPLICATED").get(1).longValue());
+    Assert.assertEquals(3, samples.get("OVER_REPLICATED").get(0).longValue());
+  }
+
   @Test
   public void testContainerIDsCanBeSampled() {
     report.incrementAndSample(
@@ -146,6 +200,35 @@
     }
   }
 
+  @Test
+  public void testDeSerializeCanHandleUnknownMetric() {
+    HddsProtos.ReplicationManagerReportProto.Builder proto =
+        HddsProtos.ReplicationManagerReportProto.newBuilder();
+    proto.setTimestamp(12345);
+
+    proto.addStat(HddsProtos.KeyIntValue.newBuilder()
+        .setKey("unknownValue")
+        .setValue(15)
+        .build());
+
+    proto.addStat(HddsProtos.KeyIntValue.newBuilder()
+        .setKey(ReplicationManagerReport.HealthState.UNDER_REPLICATED
+            .toString())
+        .setValue(20)
+        .build());
+
+    HddsProtos.KeyContainerIDList.Builder sample
+        = HddsProtos.KeyContainerIDList.newBuilder();
+    sample.setKey("unknownValue");
+    sample.addContainer(ContainerID.valueOf(1).getProtobuf());
+    proto.addStatSample(sample.build());
+    // Ensure no exception is thrown
+    ReplicationManagerReport newReport =
+        ReplicationManagerReport.fromProtobuf(proto.build());
+    Assert.assertEquals(20, newReport.getStat(
+        ReplicationManagerReport.HealthState.UNDER_REPLICATED));
+  }
+
   @Test(expected = IllegalStateException.class)
   public void testStatCannotBeSetTwice() {
     report.setStat(HddsProtos.LifeCycleState.CLOSED.toString(), 10);
diff --git a/hadoop-hdds/docs/pom.xml b/hadoop-hdds/docs/pom.xml
index fded922..9a6a330 100644
--- a/hadoop-hdds/docs/pom.xml
+++ b/hadoop-hdds/docs/pom.xml
@@ -28,9 +28,10 @@
   <name>Apache Ozone/HDDS Documentation</name>
   <packaging>jar</packaging>
 
-  <dependencies>
+  <properties>
+    <skipDocs>false</skipDocs>
+  </properties>
 
-  </dependencies>
   <build>
     <plugins>
       <plugin>
@@ -42,11 +43,12 @@
               <goal>exec</goal>
             </goals>
             <phase>compile</phase>
+            <configuration>
+              <executable>../../hadoop-ozone/dev-support/checks/docs.sh</executable>
+              <skip>${skipDocs}</skip>
+            </configuration>
           </execution>
         </executions>
-        <configuration>
-          <executable>dev-support/bin/generate-site.sh</executable>
-        </configuration>
       </plugin>
       <plugin>
         <groupId>org.apache.rat</groupId>
diff --git a/hadoop-hdds/pom.xml b/hadoop-hdds/pom.xml
index 35248b7..d981032 100644
--- a/hadoop-hdds/pom.xml
+++ b/hadoop-hdds/pom.xml
@@ -109,6 +109,12 @@
 
       <dependency>
         <groupId>org.apache.ozone</groupId>
+        <artifactId>hdds-erasurecode</artifactId>
+        <version>${hdds.version}</version>
+      </dependency>
+
+      <dependency>
+        <groupId>org.apache.ozone</groupId>
         <artifactId>hdds-client</artifactId>
         <version>${hdds.version}</version>
       </dependency>
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancer.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancer.java
index 2e7b04f..09d289e 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancer.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancer.java
@@ -461,62 +461,44 @@
    */
   private void checkIterationMoveResults(Set<DatanodeDetails> selectedTargets) {
     this.countDatanodesInvolvedPerIteration = 0;
-    this.sizeMovedPerIteration = 0;
-    for (Map.Entry<ContainerMoveSelection,
-            CompletableFuture<ReplicationManager.MoveResult>>
-        futureEntry : moveSelectionToFutureMap.entrySet()) {
-      ContainerMoveSelection moveSelection = futureEntry.getKey();
-      CompletableFuture<ReplicationManager.MoveResult> future =
-          futureEntry.getValue();
-      try {
-        ReplicationManager.MoveResult result = future.get(
-            config.getMoveTimeout().toMillis(), TimeUnit.MILLISECONDS);
-        if (result == ReplicationManager.MoveResult.COMPLETED) {
-          try {
-            ContainerInfo container =
-                containerManager.getContainer(moveSelection.getContainerID());
-            this.sizeMovedPerIteration += container.getUsedBytes();
-            metrics.incrementNumContainerMovesInLatestIteration(1);
-            LOG.info("Container move completed for container {} to target {}",
-                container.containerID(),
-                moveSelection.getTargetNode().getUuidString());
-          } catch (ContainerNotFoundException e) {
-            LOG.warn("Could not find Container {} while " +
-                    "checking move results in ContainerBalancer",
-                moveSelection.getContainerID(), e);
-          }
-        } else {
-          if (LOG.isDebugEnabled()) {
-            LOG.debug("Container move for container {} to target {} failed: {}",
-                moveSelection.getContainerID(),
-                moveSelection.getTargetNode().getUuidString(), result);
-          }
-        }
-      } catch (InterruptedException e) {
-        LOG.warn("Interrupted while waiting for container move result for " +
-                "container {}.",
-            moveSelection.getContainerID(), e);
-        Thread.currentThread().interrupt();
-      } catch (ExecutionException e) {
-        LOG.warn("Container move for container {} completed exceptionally.",
-            moveSelection.getContainerID(), e);
-      } catch (TimeoutException e) {
-        LOG.warn("Container move for container {} timed out.",
-            moveSelection.getContainerID(), e);
-      }
+
+    CompletableFuture<Void> allFuturesResult = CompletableFuture.allOf(
+        moveSelectionToFutureMap.values()
+            .toArray(new CompletableFuture[moveSelectionToFutureMap.size()]));
+    try {
+      allFuturesResult.get(config.getMoveTimeout().toMillis(),
+          TimeUnit.MILLISECONDS);
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+    } catch (TimeoutException e) {
+      long timeoutCounts = moveSelectionToFutureMap.entrySet().stream()
+          .filter(entry -> !entry.getValue().isDone())
+          .peek(entry -> {
+            LOG.warn("Container move canceled for container {} to target {} " +
+                "due to timeout.", entry.getKey().getContainerID(),
+                entry.getKey().getTargetNode().getUuidString());
+            entry.getValue().cancel(true);
+          }).count();
+      LOG.warn("{} Container moves are canceled.", timeoutCounts);
+      metrics.incrementNumContainerMovesTimeoutInLatestIteration(timeoutCounts);
+    } catch (ExecutionException e) {
+      LOG.error("Got exception while checkIterationMoveResults", e);
     }
+
     countDatanodesInvolvedPerIteration =
         sourceToTargetMap.size() + selectedTargets.size();
     metrics.incrementNumDatanodesInvolvedInLatestIteration(
         countDatanodesInvolvedPerIteration);
-    sizeMovedPerIteration /= OzoneConsts.GB;
-    metrics.incrementDataSizeMovedGBInLatestIteration(sizeMovedPerIteration);
-    metrics.incrementNumContainerMoves(
-        metrics.getNumContainerMovesInLatestIteration());
-    metrics.incrementDataSizeMovedGB(sizeMovedPerIteration);
+    metrics.incrementNumContainerMovesCompleted(
+        metrics.getNumContainerMovesCompletedInLatestIteration());
+    metrics.incrementNumContainerMovesTimeout(
+        metrics.getNumContainerMovesTimeoutInLatestIteration());
+    metrics.incrementDataSizeMovedGB(
+        metrics.getDataSizeMovedGBInLatestIteration());
     LOG.info("Number of datanodes involved in this iteration: {}. Size moved " +
             "in this iteration: {}GB.",
-        countDatanodesInvolvedPerIteration, sizeMovedPerIteration);
+        countDatanodesInvolvedPerIteration,
+        metrics.getDataSizeMovedGBInLatestIteration());
   }
 
   /**
@@ -603,25 +585,49 @@
    */
   private boolean moveContainer(DatanodeDetails source,
                                 ContainerMoveSelection moveSelection) {
-    ContainerID container = moveSelection.getContainerID();
+    ContainerID containerID = moveSelection.getContainerID();
     CompletableFuture<ReplicationManager.MoveResult> future;
     try {
+      ContainerInfo containerInfo = containerManager.getContainer(containerID);
       future = replicationManager
-          .move(container, source, moveSelection.getTargetNode());
+          .move(containerID, source, moveSelection.getTargetNode())
+          .whenComplete((result, ex) -> {
+            if (ex != null) {
+              LOG.info("Container move for container {} from source {} to " +
+                      "target {} failed with exceptions {}",
+                  containerID.toString(),
+                  source.getUuidString(),
+                  moveSelection.getTargetNode().getUuidString(), ex);
+            } else {
+              if (result == ReplicationManager.MoveResult.COMPLETED) {
+                metrics.incrementDataSizeMovedGBInLatestIteration(
+                    containerInfo.getUsedBytes() / OzoneConsts.GB);
+                metrics.incrementNumContainerMovesCompletedInLatestIteration(1);
+                if (LOG.isDebugEnabled()) {
+                  LOG.debug(
+                      "Container move completed for container {} to target {}",
+                      containerID,
+                      moveSelection.getTargetNode().getUuidString());
+                }
+              } else {
+                LOG.warn(
+                    "Container move for container {} to target {} failed: {}",
+                    moveSelection.getContainerID(),
+                    moveSelection.getTargetNode().getUuidString(), result);
+              }
+            }
+          });
     } catch (ContainerNotFoundException e) {
-      LOG.warn("Could not find Container {} for container move", container, e);
+      LOG.warn("Could not find Container {} for container move",
+          containerID, e);
       return false;
     } catch (NodeNotFoundException e) {
-      LOG.warn("Container move failed for container {}", container, e);
+      LOG.warn("Container move failed for container {}", containerID, e);
       return false;
     }
+
     if (future.isDone()) {
       if (future.isCompletedExceptionally()) {
-        LOG.info("Container move for container {} from source {} to target {}" +
-                "failed with exceptions",
-            container.toString(),
-            source.getUuidString(),
-            moveSelection.getTargetNode().getUuidString());
         return false;
       } else {
         ReplicationManager.MoveResult result = future.join();
@@ -785,7 +791,8 @@
     this.countDatanodesInvolvedPerIteration = 0;
     this.sizeMovedPerIteration = 0;
     metrics.resetDataSizeMovedGBInLatestIteration();
-    metrics.resetNumContainerMovesInLatestIteration();
+    metrics.resetNumContainerMovesCompletedInLatestIteration();
+    metrics.resetNumContainerMovesTimeoutInLatestIteration();
     metrics.resetNumDatanodesInvolvedInLatestIteration();
     metrics.resetDataSizeUnbalancedGB();
     metrics.resetNumDatanodesUnbalanced();
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerMetrics.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerMetrics.java
index f40dd44..3a7ce49 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerMetrics.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerMetrics.java
@@ -40,9 +40,13 @@
       " in the latest iteration.")
   private MutableCounterLong dataSizeMovedGBInLatestIteration;
 
-  @Metric(about = "Number of container moves performed by Container Balancer " +
-      "in the latest iteration.")
-  private MutableCounterLong numContainerMovesInLatestIteration;
+  @Metric(about = "Number of completed container moves performed by " +
+      "Container Balancer in the latest iteration.")
+  private MutableCounterLong numContainerMovesCompletedInLatestIteration;
+
+  @Metric(about = "Number of timeout container moves performed by " +
+      "Container Balancer in the latest iteration.")
+  private MutableCounterLong numContainerMovesTimeoutInLatestIteration;
 
   @Metric(about = "Number of iterations that Container Balancer has run for.")
   private MutableCounterLong numIterations;
@@ -57,9 +61,13 @@
   @Metric(about = "Number of unbalanced datanodes.")
   private MutableCounterLong numDatanodesUnbalanced;
 
-  @Metric(about = "Total number of container moves across all iterations of " +
-      "Container Balancer.")
-  private MutableCounterLong numContainerMoves;
+  @Metric(about = "Total number of completed container moves across all " +
+      "iterations of Container Balancer.")
+  private MutableCounterLong numContainerMovesCompleted;
+
+  @Metric(about = "Total number of timeout container moves across " +
+      "all iterations of Container Balancer.")
+  private MutableCounterLong numContainerMovesTimeout;
 
   @Metric(about = "Total data size in GB moved across all iterations of " +
       "Container Balancer.")
@@ -104,17 +112,37 @@
    * latest iteration.
    * @return number of container moves
    */
-  public long getNumContainerMovesInLatestIteration() {
-    return numContainerMovesInLatestIteration.value();
+  public long getNumContainerMovesCompletedInLatestIteration() {
+    return numContainerMovesCompletedInLatestIteration.value();
   }
 
-  public void incrementNumContainerMovesInLatestIteration(long valueToAdd) {
-    this.numContainerMovesInLatestIteration.incr(valueToAdd);
+  public void incrementNumContainerMovesCompletedInLatestIteration(
+      long valueToAdd) {
+    this.numContainerMovesCompletedInLatestIteration.incr(valueToAdd);
   }
 
-  public void resetNumContainerMovesInLatestIteration() {
-    numContainerMovesInLatestIteration.incr(
-        -getNumContainerMovesInLatestIteration());
+  public void resetNumContainerMovesCompletedInLatestIteration() {
+    numContainerMovesCompletedInLatestIteration.incr(
+        -getNumContainerMovesCompletedInLatestIteration());
+  }
+
+  /**
+   * Gets the number of timeout container moves performed by
+   * Container Balancer in the latest iteration.
+   * @return number of timeout container moves
+   */
+  public long getNumContainerMovesTimeoutInLatestIteration() {
+    return numContainerMovesTimeoutInLatestIteration.value();
+  }
+
+  public void incrementNumContainerMovesTimeoutInLatestIteration(
+      long valueToAdd) {
+    this.numContainerMovesTimeoutInLatestIteration.incr(valueToAdd);
+  }
+
+  public void resetNumContainerMovesTimeoutInLatestIteration() {
+    numContainerMovesTimeoutInLatestIteration.incr(
+        -getNumContainerMovesTimeoutInLatestIteration());
   }
 
   /**
@@ -179,12 +207,20 @@
     numDatanodesUnbalanced.incr(-getNumDatanodesUnbalanced());
   }
 
-  public long getNumContainerMoves() {
-    return numContainerMoves.value();
+  public long getNumContainerMovesCompleted() {
+    return numContainerMovesCompleted.value();
   }
 
-  public void incrementNumContainerMoves(long valueToAdd) {
-    numContainerMoves.incr(valueToAdd);
+  public void incrementNumContainerMovesCompleted(long valueToAdd) {
+    numContainerMovesCompleted.incr(valueToAdd);
+  }
+
+  public long getNumContainerMovesTimeout() {
+    return numContainerMovesTimeout.value();
+  }
+
+  public void incrementNumContainerMovesTimeout(long valueToAdd) {
+    numContainerMovesTimeout.incr(valueToAdd);
   }
 
   public long getDataSizeMovedGB() {
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/BackgroundPipelineCreator.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/BackgroundPipelineCreator.java
index 6fed1e2..c9eb683 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/BackgroundPipelineCreator.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/BackgroundPipelineCreator.java
@@ -29,7 +29,6 @@
 import org.apache.hadoop.hdds.scm.ScmConfigKeys;
 import org.apache.hadoop.hdds.scm.ha.SCMContext;
 import org.apache.hadoop.hdds.scm.ha.SCMService;
-import org.apache.hadoop.hdds.scm.ha.SCMServiceManager;
 import org.apache.hadoop.ozone.OzoneConfigKeys;
 import org.apache.hadoop.util.Time;
 import org.slf4j.Logger;
@@ -88,9 +87,7 @@
 
 
   BackgroundPipelineCreator(PipelineManager pipelineManager,
-                              ConfigurationSource conf,
-                              SCMServiceManager serviceManager,
-                              SCMContext scmContext) {
+      ConfigurationSource conf, SCMContext scmContext) {
     this.pipelineManager = pipelineManager;
     this.conf = conf;
     this.scmContext = scmContext;
@@ -109,9 +106,6 @@
         ScmConfigKeys.OZONE_SCM_PIPELINE_CREATION_INTERVAL_DEFAULT,
         TimeUnit.MILLISECONDS);
 
-    // register BackgroundPipelineCreator to SCMServiceManager
-    serviceManager.register(this);
-
     // start RatisPipelineUtilsThread
     start();
   }
@@ -223,13 +217,6 @@
         continue;
       }
       list.add(replicationConfig);
-      if (!pipelineManager.getSafeModeStatus()) {
-        try {
-          pipelineManager.scrubPipeline(replicationConfig);
-        } catch (IOException e) {
-          LOG.error("Error while scrubbing pipelines.", e);
-        }
-      }
     }
 
     LoopingIterator it = new LoopingIterator(list);
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/BackgroundPipelineScrubber.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/BackgroundPipelineScrubber.java
new file mode 100644
index 0000000..2063ac3
--- /dev/null
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/BackgroundPipelineScrubber.java
@@ -0,0 +1,172 @@
+/*
+ * 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.hdds.scm.pipeline;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.hadoop.hdds.HddsConfigKeys;
+import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.scm.ScmConfigKeys;
+import org.apache.hadoop.hdds.scm.ha.SCMContext;
+import org.apache.hadoop.hdds.scm.ha.SCMService;
+import org.apache.hadoop.util.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * Background service to clean up pipelines with following conditions.
+ * - CLOSED
+ * - ALLOCATED for too long
+ */
+public class BackgroundPipelineScrubber implements SCMService {
+
+  private static final Logger LOG =
+      LoggerFactory.getLogger(BackgroundPipelineScrubber.class);
+
+  private static final String THREAD_NAME = "PipelineScrubberThread";
+
+  private final PipelineManager pipelineManager;
+  private final ConfigurationSource conf;
+  private final SCMContext scmContext;
+
+  private final Lock serviceLock = new ReentrantLock();
+  private ServiceStatus serviceStatus = ServiceStatus.PAUSING;
+
+  private final AtomicBoolean running = new AtomicBoolean(false);
+  private Thread scrubThread;
+  private final long intervalInMillis;
+  private final long waitTimeInMillis;
+  private long lastTimeToBeReadyInMillis = 0;
+
+  public BackgroundPipelineScrubber(PipelineManager pipelineManager,
+      ConfigurationSource conf, SCMContext scmContext) {
+    this.pipelineManager = pipelineManager;
+    this.conf = conf;
+    this.scmContext = scmContext;
+
+    this.intervalInMillis = conf.getTimeDuration(
+        ScmConfigKeys.OZONE_SCM_PIPELINE_SCRUB_INTERVAL,
+        ScmConfigKeys.OZONE_SCM_PIPELINE_SCRUB_INTERVAL_DEFAULT,
+        TimeUnit.MILLISECONDS);
+    this.waitTimeInMillis = conf.getTimeDuration(
+        HddsConfigKeys.HDDS_SCM_WAIT_TIME_AFTER_SAFE_MODE_EXIT,
+        HddsConfigKeys.HDDS_SCM_WAIT_TIME_AFTER_SAFE_MODE_EXIT_DEFAULT,
+        TimeUnit.MILLISECONDS);
+
+    start();
+  }
+
+  @Override
+  public void notifyStatusChanged() {
+    serviceLock.lock();
+    try {
+      if (scmContext.isLeaderReady() && !scmContext.isInSafeMode()) {
+        if (serviceStatus != ServiceStatus.RUNNING) {
+          LOG.info("Service {} transitions to RUNNING.", getServiceName());
+          serviceStatus = ServiceStatus.RUNNING;
+          lastTimeToBeReadyInMillis = Time.monotonicNow();
+        }
+      } else {
+        if (serviceStatus != ServiceStatus.PAUSING) {
+          LOG.info("Service {} transitions to PAUSING.", getServiceName());
+          serviceStatus = ServiceStatus.PAUSING;
+        }
+      }
+    } finally {
+      serviceLock.unlock();
+    }
+  }
+
+  @Override
+  public boolean shouldRun() {
+    serviceLock.lock();
+    try {
+      // If safe mode is off, then this SCMService starts to run with a delay.
+      return serviceStatus == ServiceStatus.RUNNING &&
+          Time.monotonicNow() - lastTimeToBeReadyInMillis >= waitTimeInMillis;
+    } finally {
+      serviceLock.unlock();
+    }
+  }
+
+  @Override
+  public String getServiceName() {
+    return BackgroundPipelineScrubber.class.getSimpleName();
+  }
+
+  @Override
+  public void start() {
+    if (!running.compareAndSet(false, true)) {
+      LOG.info("Pipeline Scrubber Service is already running, skip start.");
+      return;
+    }
+    LOG.info("Starting Pipeline Scrubber Service.");
+
+    scrubThread = new Thread(this::run);
+    scrubThread.setName(THREAD_NAME);
+    scrubThread.setDaemon(true);
+    scrubThread.start();
+  }
+
+  @Override
+  public void stop() {
+    synchronized (this) {
+      if (!running.compareAndSet(true, false)) {
+        LOG.info("Pipeline Scrubber Service is not running, skip stop.");
+        return;
+      }
+      notifyAll();
+    }
+    LOG.info("Stopping Pipeline Scrubber Service.");
+  }
+
+  @VisibleForTesting
+  public boolean getRunning() {
+    return running.get();
+  }
+
+  private void run() {
+    while (running.get()) {
+      try {
+        if (shouldRun()) {
+          scrubAllPipelines();
+        }
+        synchronized (this) {
+          wait(intervalInMillis);
+        }
+      } catch (InterruptedException e) {
+        LOG.warn("{} is interrupted, exit", THREAD_NAME);
+        Thread.currentThread().interrupt();
+        running.set(false);
+      }
+    }
+  }
+
+  private void scrubAllPipelines() {
+    try {
+      pipelineManager.scrubPipelines();
+    } catch (IOException e) {
+      LOG.error("Unexpected error during pipeline scrubbing", e);
+    }
+  }
+}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManager.java
index 6a50876..ffce314 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManager.java
@@ -111,8 +111,7 @@
 
   void closePipeline(Pipeline pipeline, boolean onTimeout) throws IOException;
 
-  void scrubPipeline(ReplicationConfig replicationConfig)
-      throws IOException;
+  void scrubPipelines() throws IOException;
 
   void startPipelineCreator();
 
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java
index 19b7a51..83b037c 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java
@@ -76,6 +76,7 @@
   private PipelineFactory pipelineFactory;
   private PipelineStateManager stateManager;
   private BackgroundPipelineCreator backgroundPipelineCreator;
+  private BackgroundPipelineScrubber backgroundPipelineScrubber;
   private final ConfigurationSource conf;
   private final EventPublisher eventPublisher;
   // Pipeline Manager MXBean
@@ -141,10 +142,16 @@
 
     // Create background thread.
     BackgroundPipelineCreator backgroundPipelineCreator =
-        new BackgroundPipelineCreator(
-            pipelineManager, conf, serviceManager, scmContext);
+        new BackgroundPipelineCreator(pipelineManager, conf, scmContext);
+
+    BackgroundPipelineScrubber backgroundPipelineScrubber =
+        new BackgroundPipelineScrubber(pipelineManager, conf, scmContext);
 
     pipelineManager.setBackgroundPipelineCreator(backgroundPipelineCreator);
+    pipelineManager.setBackgroundPipelineScrubber(backgroundPipelineScrubber);
+
+    serviceManager.register(backgroundPipelineCreator);
+    serviceManager.register(backgroundPipelineScrubber);
 
     return pipelineManager;
   }
@@ -408,15 +415,14 @@
    * Scrub pipelines.
    */
   @Override
-  public void scrubPipeline(ReplicationConfig config)
-      throws IOException {
+  public void scrubPipelines() throws IOException {
     Instant currentTime = Instant.now();
     Long pipelineScrubTimeoutInMills = conf.getTimeDuration(
         ScmConfigKeys.OZONE_SCM_PIPELINE_ALLOCATED_TIMEOUT,
         ScmConfigKeys.OZONE_SCM_PIPELINE_ALLOCATED_TIMEOUT_DEFAULT,
         TimeUnit.MILLISECONDS);
 
-    List<Pipeline> candidates = stateManager.getPipelines(config);
+    List<Pipeline> candidates = stateManager.getPipelines();
 
     for (Pipeline p : candidates) {
       // scrub pipelines who stay ALLOCATED for too long.
@@ -437,7 +443,6 @@
         removePipeline(p);
       }
     }
-    return;
   }
 
   /**
@@ -591,6 +596,9 @@
     if (backgroundPipelineCreator != null) {
       backgroundPipelineCreator.stop();
     }
+    if (backgroundPipelineScrubber != null) {
+      backgroundPipelineScrubber.stop();
+    }
 
     if (pmInfoBean != null) {
       MBeans.unregister(this.pmInfoBean);
@@ -639,6 +647,16 @@
     return this.backgroundPipelineCreator;
   }
 
+  private void setBackgroundPipelineScrubber(
+      BackgroundPipelineScrubber backgroundPipelineScrubber) {
+    this.backgroundPipelineScrubber = backgroundPipelineScrubber;
+  }
+
+  @VisibleForTesting
+  public BackgroundPipelineScrubber getBackgroundPipelineScrubber() {
+    return this.backgroundPipelineScrubber;
+  }
+
   @VisibleForTesting
   public PipelineFactory getPipelineFactory() {
     return pipelineFactory;
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/balancer/TestContainerBalancer.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/balancer/TestContainerBalancer.java
index 0163152..48069c5 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/balancer/TestContainerBalancer.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/balancer/TestContainerBalancer.java
@@ -642,6 +642,40 @@
     containerBalancer.stop();
   }
 
+  @Test
+  public void checkIterationResultTimeout()
+      throws NodeNotFoundException, ContainerNotFoundException {
+
+    Mockito.when(replicationManager.move(Mockito.any(ContainerID.class),
+            Mockito.any(DatanodeDetails.class),
+            Mockito.any(DatanodeDetails.class)))
+        .thenReturn(genCompletableFuture(500), genCompletableFuture(2000));
+
+    balancerConfiguration.setThreshold(10);
+    balancerConfiguration.setIterations(1);
+    balancerConfiguration.setMaxSizeEnteringTarget(10 * OzoneConsts.GB);
+    balancerConfiguration.setMaxSizeToMovePerIteration(100 * OzoneConsts.GB);
+    balancerConfiguration.setMaxDatanodesPercentageToInvolvePerIteration(100);
+    balancerConfiguration.setMoveTimeout(Duration.ofMillis(1000));
+
+    startBalancer(balancerConfiguration);
+    sleepWhileBalancing(2000);
+
+    /*
+    According to the setup and configurations, this iteration's result should
+    be ITERATION_COMPLETED.
+     */
+    Assert.assertEquals(ContainerBalancer.IterationResult.ITERATION_COMPLETED,
+        containerBalancer.getIterationResult());
+    Assert.assertEquals(1,
+        containerBalancer.getMetrics()
+            .getNumContainerMovesCompletedInLatestIteration());
+    Assert.assertTrue(containerBalancer.getMetrics()
+            .getNumContainerMovesTimeoutInLatestIteration() > 1);
+    containerBalancer.stop();
+
+  }
+
   /**
    * Determines unBalanced nodes, that is, over and under utilized nodes,
    * according to the generated utilization values for nodes and the threshold.
@@ -834,4 +868,16 @@
     }
   }
 
+  private CompletableFuture<ReplicationManager.MoveResult>
+      genCompletableFuture(int sleepMilSec) {
+    return CompletableFuture.supplyAsync(() -> {
+      try {
+        Thread.sleep(sleepMilSec);
+      } catch (InterruptedException e) {
+        e.printStackTrace();
+      }
+      return ReplicationManager.MoveResult.COMPLETED;
+    });
+  }
+
 }
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockPipelineManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockPipelineManager.java
index 523efc0..932d2d5 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockPipelineManager.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockPipelineManager.java
@@ -218,8 +218,7 @@
   }
 
   @Override
-  public void scrubPipeline(ReplicationConfig replicationConfig)
-      throws IOException {
+  public void scrubPipelines() {
 
   }
 
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestBackgroundPipelineScrubber.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestBackgroundPipelineScrubber.java
new file mode 100644
index 0000000..dfa5284
--- /dev/null
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestBackgroundPipelineScrubber.java
@@ -0,0 +1,99 @@
+/*
+ * 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.hdds.scm.pipeline;
+
+import org.apache.hadoop.hdds.HddsConfigKeys;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.scm.ha.SCMContext;
+import org.apache.hadoop.hdds.scm.safemode.SCMSafeModeManager.SafeModeStatus;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.IOException;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
+
+/**
+ * Test for {@link BackgroundPipelineScrubber}.
+ */
+public class TestBackgroundPipelineScrubber {
+
+  private BackgroundPipelineScrubber scrubber;
+  private SCMContext scmContext;
+  private PipelineManager pipelineManager;
+  private OzoneConfiguration conf;
+
+  @Before
+  public void setup() throws IOException {
+    this.scmContext = SCMContext.emptyContext();
+    this.pipelineManager = mock(PipelineManager.class);
+    doNothing().when(pipelineManager).scrubPipelines();
+
+    // no initial delay after exit safe mode
+    this.conf = new OzoneConfiguration();
+    conf.set(HddsConfigKeys.HDDS_SCM_WAIT_TIME_AFTER_SAFE_MODE_EXIT, "0ms");
+
+    this.scrubber = new BackgroundPipelineScrubber(pipelineManager, conf,
+        scmContext);
+  }
+
+  @After
+  public void teardown() throws IOException {
+    scrubber.stop();
+  }
+
+  @Test
+  public void testStop() {
+    assertTrue(scrubber.getRunning());
+    scrubber.stop();
+    assertFalse(scrubber.getRunning());
+  }
+
+  @Test
+  public void testNotifyStatusChanged() {
+    // init at PAUSING
+    assertFalse(scrubber.shouldRun());
+
+    // out of safe mode, PAUSING -> RUNNING
+    scrubber.notifyStatusChanged();
+    assertTrue(scrubber.shouldRun());
+
+    // go into safe mode, RUNNING -> PAUSING
+    scmContext.updateSafeModeStatus(new SafeModeStatus(true, true));
+    scrubber.notifyStatusChanged();
+    assertFalse(scrubber.shouldRun());
+  }
+
+  @Test
+  public void testRun() throws IOException {
+    assertFalse(scrubber.shouldRun());
+    // kick a run
+    synchronized (scrubber) {
+      scrubber.notifyStatusChanged();
+      assertTrue(scrubber.shouldRun());
+      scrubber.notifyAll();
+    }
+    verify(pipelineManager, timeout(3000).times(1)).scrubPipelines();
+  }
+}
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelineManagerImpl.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelineManagerImpl.java
index fd29929..64ef02e 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelineManagerImpl.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelineManagerImpl.java
@@ -535,7 +535,7 @@
   }
 
   @Test
-  public void testScrubPipeline() throws Exception {
+  public void testScrubPipelines() throws Exception {
     // No timeout for pipeline scrubber.
     conf.setTimeDuration(
         OZONE_SCM_PIPELINE_ALLOCATED_TIMEOUT, -1,
@@ -543,33 +543,50 @@
 
     PipelineManagerImpl pipelineManager = createPipelineManager(true);
     pipelineManager.setScmContext(scmContext);
-    Pipeline pipeline = pipelineManager
+    Pipeline allocatedPipeline = pipelineManager
         .createPipeline(RatisReplicationConfig
             .getInstance(ReplicationFactor.THREE));
     // At this point, pipeline is not at OPEN stage.
     Assert.assertEquals(Pipeline.PipelineState.ALLOCATED,
-        pipeline.getPipelineState());
+        allocatedPipeline.getPipelineState());
 
     // pipeline should be seen in pipelineManager as ALLOCATED.
     Assert.assertTrue(pipelineManager
         .getPipelines(RatisReplicationConfig
             .getInstance(ReplicationFactor.THREE),
-            Pipeline.PipelineState.ALLOCATED).contains(pipeline));
-    pipelineManager
-        .scrubPipeline(RatisReplicationConfig
-            .getInstance(ReplicationFactor.THREE));
+            Pipeline.PipelineState.ALLOCATED).contains(allocatedPipeline));
 
-    // pipeline should be scrubbed.
+    Pipeline closedPipeline = pipelineManager
+        .createPipeline(RatisReplicationConfig
+            .getInstance(ReplicationFactor.THREE));
+    pipelineManager.openPipeline(closedPipeline.getId());
+    pipelineManager.closePipeline(closedPipeline, true);
+
+    // pipeline should be seen in pipelineManager as CLOSED.
+    Assert.assertTrue(pipelineManager
+        .getPipelines(RatisReplicationConfig
+                .getInstance(ReplicationFactor.THREE),
+            Pipeline.PipelineState.CLOSED).contains(closedPipeline));
+
+    pipelineManager.scrubPipelines();
+
+    // The allocatedPipeline should be scrubbed.
     Assert.assertFalse(pipelineManager
         .getPipelines(RatisReplicationConfig
             .getInstance(ReplicationFactor.THREE),
-            Pipeline.PipelineState.ALLOCATED).contains(pipeline));
+            Pipeline.PipelineState.ALLOCATED).contains(allocatedPipeline));
+
+    // The closedPipeline should be scrubbed.
+    Assert.assertFalse(pipelineManager
+        .getPipelines(RatisReplicationConfig
+                .getInstance(ReplicationFactor.THREE),
+            Pipeline.PipelineState.CLOSED).contains(closedPipeline));
 
     pipelineManager.close();
   }
 
   @Test
-  public void testScrubPipelineShouldFailOnFollower() throws Exception {
+  public void testScrubPipelinesShouldFailOnFollower() throws Exception {
     // No timeout for pipeline scrubber.
     conf.setTimeDuration(
         OZONE_SCM_PIPELINE_ALLOCATED_TIMEOUT, -1,
@@ -595,9 +612,7 @@
     ((SCMHAManagerStub) pipelineManager.getScmhaManager()).setIsLeader(false);
 
     try {
-      pipelineManager
-          .scrubPipeline(RatisReplicationConfig
-              .getInstance(ReplicationFactor.THREE));
+      pipelineManager.scrubPipelines();
     } catch (NotLeaderException ex) {
       pipelineManager.close();
       return;
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/safemode/TestSCMSafeModeManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/safemode/TestSCMSafeModeManager.java
index 1062274..823c2dd 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/safemode/TestSCMSafeModeManager.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/safemode/TestSCMSafeModeManager.java
@@ -354,6 +354,7 @@
     pipelineManager.setPipelineProvider(HddsProtos.ReplicationType.RATIS,
         mockRatisProvider);
     pipelineManager.getBackgroundPipelineCreator().stop();
+    pipelineManager.getBackgroundPipelineScrubber().stop();
 
     for (int i = 0; i < pipelineCount; i++) {
       // Create pipeline
diff --git a/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/container/InfoSubcommand.java b/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/container/InfoSubcommand.java
index 37bea18..25ed0e5 100644
--- a/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/container/InfoSubcommand.java
+++ b/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/container/InfoSubcommand.java
@@ -63,11 +63,6 @@
       description = "Format output as JSON")
   private boolean json;
 
-  @CommandLine.Option(names = { "--replicas" },
-      defaultValue = "false",
-      description = "Adds replica related details")
-  private boolean addReplicaDetails;
-
   @Parameters(description = "Decimal id of the container.")
   private long containerID;
 
@@ -108,7 +103,7 @@
       LOG.info("Datanodes: [{}]", machinesStr);
 
       // Print the replica details if available
-      if (addReplicaDetails && replicas != null) {
+      if (replicas != null) {
         String replicaStr = replicas.stream().map(
             InfoSubcommand::buildReplicaDetails)
             .collect(Collectors.joining(",\n"));
diff --git a/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/container/ReportSubcommand.java b/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/container/ReportSubcommand.java
index 89b1a11..554316c 100644
--- a/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/container/ReportSubcommand.java
+++ b/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/container/ReportSubcommand.java
@@ -23,6 +23,7 @@
 import org.apache.hadoop.hdds.scm.client.ScmClient;
 import org.apache.hadoop.hdds.scm.container.ContainerID;
 import org.apache.hadoop.hdds.scm.container.ReplicationManagerReport;
+import org.apache.hadoop.hdds.server.JsonUtils;
 import picocli.CommandLine;
 
 import java.io.IOException;
@@ -43,9 +44,20 @@
   @CommandLine.Spec
   private CommandLine.Model.CommandSpec spec;
 
+  @CommandLine.Option(names = { "--json" },
+      defaultValue = "false",
+      description = "Format output as JSON")
+  private boolean json;
+
   @Override
   public void execute(ScmClient scmClient) throws IOException {
     ReplicationManagerReport report = scmClient.getReplicationManagerReport();
+
+    if (json) {
+      output(JsonUtils.toJsonStringWithDefaultPrettyPrinter(report));
+      return;
+    }
+
     outputHeader(report.getReportTimeStamp());
     blankLine();
     outputContainerStats(report);
diff --git a/hadoop-hdds/tools/src/test/java/org/apache/hadoop/hdds/scm/cli/container/TestInfoSubCommand.java b/hadoop-hdds/tools/src/test/java/org/apache/hadoop/hdds/scm/cli/container/TestInfoSubCommand.java
index c604019..10b5758 100644
--- a/hadoop-hdds/tools/src/test/java/org/apache/hadoop/hdds/scm/cli/container/TestInfoSubCommand.java
+++ b/hadoop-hdds/tools/src/test/java/org/apache/hadoop/hdds/scm/cli/container/TestInfoSubCommand.java
@@ -97,7 +97,7 @@
         .thenReturn(getReplicas(includeIndex));
     cmd = new InfoSubcommand();
     CommandLine c = new CommandLine(cmd);
-    c.parseArgs("1", "--replicas");
+    c.parseArgs("1");
     cmd.execute(scmClient);
 
     // Ensure we have a line for Replicas:
diff --git a/hadoop-hdds/tools/src/test/java/org/apache/hadoop/hdds/scm/cli/container/TestReportSubCommand.java b/hadoop-hdds/tools/src/test/java/org/apache/hadoop/hdds/scm/cli/container/TestReportSubCommand.java
index be0e2c8..df8dd95 100644
--- a/hadoop-hdds/tools/src/test/java/org/apache/hadoop/hdds/scm/cli/container/TestReportSubCommand.java
+++ b/hadoop-hdds/tools/src/test/java/org/apache/hadoop/hdds/scm/cli/container/TestReportSubCommand.java
@@ -17,14 +17,18 @@
  */
 package org.apache.hadoop.hdds.scm.cli.container;
 
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 import org.apache.hadoop.hdds.scm.client.ScmClient;
 import org.apache.hadoop.hdds.scm.container.ContainerID;
 import org.apache.hadoop.hdds.scm.container.ReplicationManagerReport;
 import org.junit.After;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 import org.mockito.Mockito;
+import picocli.CommandLine;
 
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
@@ -88,6 +92,26 @@
   }
 
   @Test
+  public void testValidJsonOutput() throws IOException {
+    // More complete testing of the Report JSON output is in
+    // TestReplicationManagerReport.
+    ScmClient scmClient = mock(ScmClient.class);
+    Mockito.when(scmClient.getReplicationManagerReport())
+        .thenAnswer(invocation -> new ReplicationManagerReport());
+
+    CommandLine c = new CommandLine(cmd);
+    c.parseArgs("--json");
+    cmd.execute(scmClient);
+
+    ObjectMapper mapper = new ObjectMapper();
+    JsonNode json = mapper.readTree(outContent.toString("UTF-8"));
+
+    Assert.assertTrue(json.get("reportTimeStamp") != null);
+    Assert.assertTrue(json.get("stats") != null);
+    Assert.assertTrue(json.get("samples") != null);
+  }
+
+  @Test
   public void testCorrectValuesAppearInReport() throws IOException {
     ScmClient scmClient = mock(ScmClient.class);
     Mockito.when(scmClient.getReplicationManagerReport())
diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyLocationInfo.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyLocationInfo.java
index 9df7518..9ff1ca9 100644
--- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyLocationInfo.java
+++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyLocationInfo.java
@@ -19,156 +19,72 @@
 import org.apache.hadoop.hdds.client.BlockID;
 import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
 import org.apache.hadoop.hdds.scm.pipeline.UnknownPipelineStateException;
+import org.apache.hadoop.hdds.scm.storage.BlockLocationInfo;
 import org.apache.hadoop.hdds.security.token.OzoneBlockTokenIdentifier;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.KeyLocation;
 import org.apache.hadoop.ozone.protocolPB.OzonePBHelper;
 import org.apache.hadoop.security.token.Token;
 
-import java.util.Objects;
-
 /**
  * One key can be too huge to fit in one container. In which case it gets split
  * into a number of subkeys. This class represents one such subkey instance.
  */
-public final class OmKeyLocationInfo {
-  private final BlockID blockID;
-  // the id of this subkey in all the subkeys.
-  private long length;
-  private final long offset;
-  // Block token, required for client authentication when security is enabled.
-  private Token<OzoneBlockTokenIdentifier> token;
-  // the version number indicating when this block was added
-  private long createVersion;
+public final class OmKeyLocationInfo extends BlockLocationInfo {
 
-  private Pipeline pipeline;
-
-  // PartNumber is set for Multipart upload Keys.
-  private int partNumber = -1;
-
-  private OmKeyLocationInfo(BlockID blockID, Pipeline pipeline, long length,
-                            long offset, int partNumber) {
-    this.blockID = blockID;
-    this.pipeline = pipeline;
-    this.length = length;
-    this.offset = offset;
-    this.partNumber = partNumber;
-  }
-
-  private OmKeyLocationInfo(BlockID blockID, Pipeline pipeline, long length,
-      long offset, Token<OzoneBlockTokenIdentifier> token, int partNumber) {
-    this.blockID = blockID;
-    this.pipeline = pipeline;
-    this.length = length;
-    this.offset = offset;
-    this.token = token;
-    this.partNumber = partNumber;
-  }
-
-  public void setCreateVersion(long version) {
-    createVersion = version;
-  }
-
-  public long getCreateVersion() {
-    return createVersion;
-  }
-
-  public BlockID getBlockID() {
-    return blockID;
-  }
-
-  public long getContainerID() {
-    return blockID.getContainerID();
-  }
-
-  public long getLocalID() {
-    return blockID.getLocalID();
-  }
-
-  public Pipeline getPipeline() {
-    return pipeline;
-  }
-
-  public long getLength() {
-    return length;
-  }
-
-  public void setLength(long length) {
-    this.length = length;
-  }
-
-  public long getOffset() {
-    return offset;
-  }
-
-  public long getBlockCommitSequenceId() {
-    return blockID.getBlockCommitSequenceId();
-  }
-
-  public Token<OzoneBlockTokenIdentifier> getToken() {
-    return token;
-  }
-
-  public void setToken(Token<OzoneBlockTokenIdentifier> token) {
-    this.token = token;
-  }
-
-  public void setPipeline(Pipeline pipeline) {
-    this.pipeline = pipeline;
-  }
-
-  public void setPartNumber(int partNumber) {
-    this.partNumber = partNumber;
-  }
-
-  public int getPartNumber() {
-    return partNumber;
+  private OmKeyLocationInfo(Builder builder) {
+    super(builder);
   }
 
   /**
    * Builder of OmKeyLocationInfo.
    */
-  public static class Builder {
-    private BlockID blockID;
-    private long length;
-    private long offset;
-    private Token<OzoneBlockTokenIdentifier> token;
-    private Pipeline pipeline;
-    private int partNumber;
+  public static class Builder extends BlockLocationInfo.Builder {
 
-    public Builder setBlockID(BlockID blockId) {
-      this.blockID = blockId;
+    @Override
+    public Builder setBlockID(BlockID blockID) {
+      super.setBlockID(blockID);
       return this;
     }
 
-    @SuppressWarnings("checkstyle:hiddenfield")
+    @Override
     public Builder setPipeline(Pipeline pipeline) {
-      this.pipeline = pipeline;
+      super.setPipeline(pipeline);
       return this;
     }
 
-    public Builder setLength(long len) {
-      this.length = len;
+    @Override
+    public Builder setLength(long length) {
+      super.setLength(length);
       return this;
     }
 
-    public Builder setOffset(long off) {
-      this.offset = off;
+    @Override
+    public Builder setOffset(long offset) {
+      super.setOffset(offset);
       return this;
     }
 
-    public Builder setToken(Token<OzoneBlockTokenIdentifier> bToken) {
-      this.token = bToken;
+    @Override
+    public Builder setToken(Token<OzoneBlockTokenIdentifier> token) {
+      super.setToken(token);
       return this;
     }
 
-    public Builder setPartNumber(int partNum) {
-      this.partNumber = partNum;
+    @Override
+    public Builder setPartNumber(int partNumber) {
+      super.setPartNumber(partNumber);
       return this;
     }
 
+    @Override
+    public Builder setCreateVersion(long version) {
+      super.setCreateVersion(version);
+      return this;
+    }
+
+    @Override
     public OmKeyLocationInfo build() {
-      return new OmKeyLocationInfo(blockID, pipeline, length, offset, token,
-          partNumber);
+      return new OmKeyLocationInfo(this);
     }
   }
 
@@ -178,13 +94,15 @@
 
   public KeyLocation getProtobuf(boolean ignorePipeline, int clientVersion) {
     KeyLocation.Builder builder = KeyLocation.newBuilder()
-        .setBlockID(blockID.getProtobuf())
-        .setLength(length)
-        .setOffset(offset)
-        .setCreateVersion(createVersion).setPartNumber(partNumber);
+        .setBlockID(getBlockID().getProtobuf())
+        .setLength(getLength())
+        .setOffset(getOffset())
+        .setCreateVersion(getCreateVersion())
+        .setPartNumber(getPartNumber());
     if (!ignorePipeline) {
       try {
-        if (this.token != null) {
+        Token<OzoneBlockTokenIdentifier> token = getToken();
+        if (token != null) {
           builder.setToken(OzonePBHelper.protoFromToken(token));
         }
 
@@ -200,7 +118,8 @@
         // TODO: this needs to be revisited when bucket versioning
         //  implementation is handled.
 
-        if (this.pipeline != null) {
+        Pipeline pipeline = getPipeline();
+        if (pipeline != null) {
           builder.setPipeline(pipeline.getProtobufMessage(clientVersion));
         }
       } catch (UnknownPipelineStateException e) {
@@ -220,51 +139,18 @@
   }
 
   public static OmKeyLocationInfo getFromProtobuf(KeyLocation keyLocation) {
-    OmKeyLocationInfo info = new OmKeyLocationInfo(
-        BlockID.getFromProtobuf(keyLocation.getBlockID()),
-        getPipeline(keyLocation),
-        keyLocation.getLength(),
-        keyLocation.getOffset(), keyLocation.getPartNumber());
+    Builder builder = new Builder()
+        .setBlockID(BlockID.getFromProtobuf(keyLocation.getBlockID()))
+        .setLength(keyLocation.getLength())
+        .setOffset(keyLocation.getOffset())
+        .setPipeline(getPipeline(keyLocation))
+        .setCreateVersion(keyLocation.getCreateVersion())
+        .setPartNumber(keyLocation.getPartNumber());
     if (keyLocation.hasToken()) {
-      info.token = (Token<OzoneBlockTokenIdentifier>)
-              OzonePBHelper.tokenFromProto(keyLocation.getToken());
+      builder.setToken((Token<OzoneBlockTokenIdentifier>)
+          OzonePBHelper.tokenFromProto(keyLocation.getToken()));
     }
-    info.setCreateVersion(keyLocation.getCreateVersion());
-    return info;
+    return builder.build();
   }
 
-  @Override
-  public String  toString() {
-    return "{blockID={containerID=" + blockID.getContainerID() +
-        ", localID=" + blockID.getLocalID() + "}" +
-        ", length=" + length +
-        ", offset=" + offset +
-        ", token=" + token +
-        ", pipeline=" + pipeline +
-        ", createVersion=" + createVersion  + ", partNumber=" + partNumber
-        + '}';
-  }
-
-  @Override
-  public boolean equals(Object o) {
-    if (this == o) {
-      return true;
-    }
-    if (o == null || getClass() != o.getClass()) {
-      return false;
-    }
-    OmKeyLocationInfo that = (OmKeyLocationInfo) o;
-    return length == that.length &&
-        offset == that.offset &&
-        createVersion == that.createVersion &&
-        Objects.equals(blockID, that.blockID) &&
-        Objects.equals(token, that.token) &&
-        Objects.equals(pipeline, that.pipeline);
-  }
-
-  @Override
-  public int hashCode() {
-    return Objects.hash(blockID, length, offset, token, createVersion,
-        pipeline);
-  }
 }
diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/lock/OzoneManagerLock.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/lock/OzoneManagerLock.java
index 11dc5b2..f597150 100644
--- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/lock/OzoneManagerLock.java
+++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/lock/OzoneManagerLock.java
@@ -585,6 +585,10 @@
     omLockMetrics.unRegister();
   }
 
+  public OMLockMetrics getOMLockMetrics() {
+    return omLockMetrics;
+  }
+
   /**
    * Resource defined in Ozone.
    */
diff --git a/hadoop-ozone/dev-support/checks/build.sh b/hadoop-ozone/dev-support/checks/build.sh
index 26d66b8..42edeb5 100755
--- a/hadoop-ozone/dev-support/checks/build.sh
+++ b/hadoop-ozone/dev-support/checks/build.sh
@@ -17,5 +17,5 @@
 cd "$DIR/../../.." || exit 1
 
 export MAVEN_OPTS="-Xmx4096m $MAVEN_OPTS"
-mvn -V -B -Dmaven.javadoc.skip=true -DskipTests clean install "$@"
+mvn -V -B -Dmaven.javadoc.skip=true -DskipTests -DskipDocs clean install "$@"
 exit $?
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFileSystem.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFileSystem.java
index 4c4bdc6..0bbf8c9 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFileSystem.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFileSystem.java
@@ -90,6 +90,7 @@
 
 import org.apache.ozone.test.LambdaTestUtils;
 import org.apache.ozone.test.TestClock;
+import org.apache.ozone.test.tag.Flaky;
 import org.junit.After;
 import org.junit.AfterClass;
 import org.junit.Assert;
@@ -1297,6 +1298,7 @@
    * since fs.rename(src,dst,options) is enabled.
    */
   @Test
+  @Flaky("HDDS-6646")
   public void testRenameToTrashEnabled() throws Exception {
     // Create a file
     String testKeyName = "testKey1";
@@ -1326,6 +1328,7 @@
    * 2.Verify that the key gets deleted by the trash emptier.
    */
   @Test
+  @Flaky("HDDS-6645")
   public void testTrash() throws Exception {
     String testKeyName = "testKey2";
     Path path = new Path(OZONE_URI_DELIMITER, testKeyName);
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestRootedOzoneFileSystem.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestRootedOzoneFileSystem.java
index 29f3326..695d22d 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestRootedOzoneFileSystem.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestRootedOzoneFileSystem.java
@@ -61,6 +61,7 @@
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.ozone.test.GenericTestUtils;
 import org.apache.ozone.test.LambdaTestUtils;
+import org.apache.ozone.test.tag.Flaky;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Assume;
@@ -1260,6 +1261,7 @@
    * fs.rename(src, dst, options).
    */
   @Test
+  @Flaky({"HDDS-5819", "HDDS-6451"})
   public void testRenameToTrashEnabled() throws IOException {
     // Create a file
     String testKeyName = "testKey2";
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/ECStreamTestUtil.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/ECStreamTestUtil.java
index 0503dbe..f3649ea 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/ECStreamTestUtil.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/ECStreamTestUtil.java
@@ -26,10 +26,10 @@
 import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
 import org.apache.hadoop.hdds.scm.pipeline.PipelineID;
 import org.apache.hadoop.hdds.scm.storage.BlockExtendedInputStream;
+import org.apache.hadoop.hdds.scm.storage.BlockLocationInfo;
 import org.apache.hadoop.hdds.scm.storage.ByteReaderStrategy;
 import org.apache.hadoop.hdds.security.token.OzoneBlockTokenIdentifier;
 import org.apache.hadoop.ozone.client.io.BlockInputStreamFactory;
-import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
 import org.apache.hadoop.security.token.Token;
 import org.apache.ozone.erasurecode.rawcoder.RawErasureEncoder;
 import org.apache.ozone.erasurecode.rawcoder.util.CodecUtil;
@@ -57,7 +57,7 @@
   private ECStreamTestUtil() {
   }
 
-  public static OmKeyLocationInfo createKeyInfo(ReplicationConfig repConf,
+  public static BlockLocationInfo createKeyInfo(ReplicationConfig repConf,
       long blockLength, Map<DatanodeDetails, Integer> dnMap) {
 
     Pipeline pipeline = Pipeline.newBuilder()
@@ -68,7 +68,7 @@
         .setReplicationConfig(repConf)
         .build();
 
-    OmKeyLocationInfo keyInfo = new OmKeyLocationInfo.Builder()
+    BlockLocationInfo keyInfo = new BlockLocationInfo.Builder()
         .setBlockID(new BlockID(1, 1))
         .setLength(blockLength)
         .setOffset(0)
@@ -78,7 +78,7 @@
     return keyInfo;
   }
 
-  public static OmKeyLocationInfo createKeyInfo(ReplicationConfig repConf,
+  public static BlockLocationInfo createKeyInfo(ReplicationConfig repConf,
       int nodeCount, long blockLength) {
     Map<DatanodeDetails, Integer> datanodes = new HashMap<>();
     for (int i = 0; i < nodeCount; i++) {
@@ -257,7 +257,7 @@
 
     public synchronized BlockExtendedInputStream create(
         ReplicationConfig repConfig,
-        OmKeyLocationInfo blockInfo, Pipeline pipeline,
+        BlockLocationInfo blockInfo, Pipeline pipeline,
         Token<OzoneBlockTokenIdentifier> token, boolean verifyChecksum,
         XceiverClientFactory xceiverFactory,
         Function<BlockID, Pipeline> refreshFunction) {
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestECBlockInputStream.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestECBlockInputStream.java
index 17a0a6f..a9f9407 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestECBlockInputStream.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestECBlockInputStream.java
@@ -25,12 +25,12 @@
 import org.apache.hadoop.hdds.scm.XceiverClientFactory;
 import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
 import org.apache.hadoop.hdds.scm.storage.BlockExtendedInputStream;
+import org.apache.hadoop.hdds.scm.storage.BlockLocationInfo;
 import org.apache.hadoop.hdds.scm.storage.ByteReaderStrategy;
 import org.apache.hadoop.hdds.security.token.OzoneBlockTokenIdentifier;
 import org.apache.hadoop.ozone.client.io.BadDataLocationException;
 import org.apache.hadoop.ozone.client.io.BlockInputStreamFactory;
 import org.apache.hadoop.ozone.client.io.ECBlockInputStream;
-import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
 import org.apache.hadoop.security.token.Token;
 import org.junit.Assert;
 import org.junit.Before;
@@ -65,7 +65,7 @@
   @Test
   public void testSufficientLocations() {
     // EC-3-2, 5MB block, so all 3 data locations are needed
-    OmKeyLocationInfo keyInfo = ECStreamTestUtil
+    BlockLocationInfo keyInfo = ECStreamTestUtil
         .createKeyInfo(repConfig, 5, 5 * ONEMB);
     try (ECBlockInputStream ecb = new ECBlockInputStream(repConfig,
         keyInfo, true, null, null, new TestBlockInputStreamFactory())) {
@@ -116,7 +116,7 @@
   public void testCorrectBlockSizePassedToBlockStreamLessThanCell()
       throws IOException {
     ByteBuffer buf = ByteBuffer.allocate(3 * ONEMB);
-    OmKeyLocationInfo keyInfo =
+    BlockLocationInfo keyInfo =
         ECStreamTestUtil.createKeyInfo(repConfig, 5, ONEMB - 100);
 
     try (ECBlockInputStream ecb = new ECBlockInputStream(repConfig,
@@ -133,7 +133,7 @@
   public void testCorrectBlockSizePassedToBlockStreamTwoCells()
       throws IOException {
     ByteBuffer buf = ByteBuffer.allocate(3 * ONEMB);
-    OmKeyLocationInfo keyInfo =
+    BlockLocationInfo keyInfo =
         ECStreamTestUtil.createKeyInfo(repConfig, 5, ONEMB + 100);
 
     try (ECBlockInputStream ecb = new ECBlockInputStream(repConfig,
@@ -149,7 +149,7 @@
   public void testCorrectBlockSizePassedToBlockStreamThreeCells()
       throws IOException {
     ByteBuffer buf = ByteBuffer.allocate(3 * ONEMB);
-    OmKeyLocationInfo keyInfo =
+    BlockLocationInfo keyInfo =
         ECStreamTestUtil.createKeyInfo(repConfig, 5, 2 * ONEMB + 100);
 
     try (ECBlockInputStream ecb = new ECBlockInputStream(repConfig,
@@ -166,7 +166,7 @@
   public void testCorrectBlockSizePassedToBlockStreamThreeFullAndPartialStripe()
       throws IOException {
     ByteBuffer buf = ByteBuffer.allocate(3 * ONEMB);
-    OmKeyLocationInfo keyInfo =
+    BlockLocationInfo keyInfo =
         ECStreamTestUtil.createKeyInfo(repConfig, 5, 10 * ONEMB + 100);
 
     try (ECBlockInputStream ecb = new ECBlockInputStream(repConfig,
@@ -183,7 +183,7 @@
   public void testCorrectBlockSizePassedToBlockStreamSingleFullCell()
       throws IOException {
     ByteBuffer buf = ByteBuffer.allocate(3 * ONEMB);
-    OmKeyLocationInfo keyInfo =
+    BlockLocationInfo keyInfo =
         ECStreamTestUtil.createKeyInfo(repConfig, 5, ONEMB);
 
     try (ECBlockInputStream ecb = new ECBlockInputStream(repConfig,
@@ -198,7 +198,7 @@
   public void testCorrectBlockSizePassedToBlockStreamSeveralFullCells()
       throws IOException {
     ByteBuffer buf = ByteBuffer.allocate(3 * ONEMB);
-    OmKeyLocationInfo keyInfo =
+    BlockLocationInfo keyInfo =
         ECStreamTestUtil.createKeyInfo(repConfig, 5, 9 * ONEMB);
 
     try (ECBlockInputStream ecb = new ECBlockInputStream(repConfig,
@@ -213,7 +213,7 @@
 
   @Test
   public void testSimpleRead() throws IOException {
-    OmKeyLocationInfo keyInfo =
+    BlockLocationInfo keyInfo =
         ECStreamTestUtil.createKeyInfo(repConfig, 5, 5 * ONEMB);
     try (ECBlockInputStream ecb = new ECBlockInputStream(repConfig,
         keyInfo, true, null, null, streamFactory)) {
@@ -236,7 +236,7 @@
    */
   @Test
   public void testSimpleReadUnderOneChunk() throws IOException {
-    OmKeyLocationInfo keyInfo =
+    BlockLocationInfo keyInfo =
         ECStreamTestUtil.createKeyInfo(repConfig, 1, ONEMB);
     try (ECBlockInputStream ecb = new ECBlockInputStream(repConfig,
         keyInfo, true, null, null, streamFactory)) {
@@ -255,7 +255,7 @@
 
   @Test
   public void testReadPastEOF() throws IOException {
-    OmKeyLocationInfo keyInfo =
+    BlockLocationInfo keyInfo =
         ECStreamTestUtil.createKeyInfo(repConfig, 5, 50);
     try (ECBlockInputStream ecb = new ECBlockInputStream(repConfig,
         keyInfo, true, null, null, streamFactory)) {
@@ -274,7 +274,7 @@
     // EC-3-2, 5MB block, so all 3 data locations are needed
     repConfig = new ECReplicationConfig(3, 2, ECReplicationConfig.EcCodec.RS,
         100);
-    OmKeyLocationInfo keyInfo =
+    BlockLocationInfo keyInfo =
         ECStreamTestUtil.createKeyInfo(repConfig, 5, 5 * ONEMB);
     try (ECBlockInputStream ecb = new ECBlockInputStream(repConfig,
         keyInfo, true, null, null, streamFactory)) {
@@ -309,7 +309,7 @@
   public void testSeekPastBlockLength() throws IOException {
     repConfig = new ECReplicationConfig(3, 2, ECReplicationConfig.EcCodec.RS,
         ONEMB);
-    OmKeyLocationInfo keyInfo =
+    BlockLocationInfo keyInfo =
         ECStreamTestUtil.createKeyInfo(repConfig, 5, 100);
     try (ECBlockInputStream ecb = new ECBlockInputStream(repConfig,
         keyInfo, true, null, null, streamFactory)) {
@@ -321,7 +321,7 @@
   public void testSeekToLength() throws IOException {
     repConfig = new ECReplicationConfig(3, 2, ECReplicationConfig.EcCodec.RS,
         ONEMB);
-    OmKeyLocationInfo keyInfo =
+    BlockLocationInfo keyInfo =
         ECStreamTestUtil.createKeyInfo(repConfig, 5, 100);
     try (ECBlockInputStream ecb = new ECBlockInputStream(repConfig,
         keyInfo, true, null, null, streamFactory)) {
@@ -333,7 +333,7 @@
   public void testSeekToLengthZeroLengthBlock() throws IOException {
     repConfig = new ECReplicationConfig(3, 2, ECReplicationConfig.EcCodec.RS,
         ONEMB);
-    OmKeyLocationInfo keyInfo =
+    BlockLocationInfo keyInfo =
         ECStreamTestUtil.createKeyInfo(repConfig, 5, 0);
     try (ECBlockInputStream ecb = new ECBlockInputStream(repConfig,
         keyInfo, true, null, null, streamFactory)) {
@@ -347,7 +347,7 @@
   public void testSeekToValidPosition() throws IOException {
     repConfig = new ECReplicationConfig(3, 2, ECReplicationConfig.EcCodec.RS,
         ONEMB);
-    OmKeyLocationInfo keyInfo =
+    BlockLocationInfo keyInfo =
         ECStreamTestUtil.createKeyInfo(repConfig, 5, 5 * ONEMB);
     try (ECBlockInputStream ecb = new ECBlockInputStream(repConfig,
         keyInfo, true, null, null, streamFactory)) {
@@ -376,7 +376,7 @@
   public void testErrorReadingBlockReportsBadLocation() throws IOException {
     repConfig = new ECReplicationConfig(3, 2, ECReplicationConfig.EcCodec.RS,
         ONEMB);
-    OmKeyLocationInfo keyInfo =
+    BlockLocationInfo keyInfo =
         ECStreamTestUtil.createKeyInfo(repConfig, 5, 5 * ONEMB);
     try (ECBlockInputStream ecb = new ECBlockInputStream(repConfig,
         keyInfo, true, null, null, streamFactory)) {
@@ -417,7 +417,7 @@
     }
 
     public synchronized BlockExtendedInputStream create(
-        ReplicationConfig repConfig, OmKeyLocationInfo blockInfo,
+        ReplicationConfig repConfig, BlockLocationInfo blockInfo,
         Pipeline pipeline, Token<OzoneBlockTokenIdentifier> token,
         boolean verifyChecksum, XceiverClientFactory xceiverFactory,
         Function<BlockID, Pipeline> refreshFunction) {
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestECBlockInputStreamProxy.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestECBlockInputStreamProxy.java
index fd4e8ad..a45ab43 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestECBlockInputStreamProxy.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestECBlockInputStreamProxy.java
@@ -24,10 +24,10 @@
 import org.apache.hadoop.hdds.scm.XceiverClientFactory;
 import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
 import org.apache.hadoop.hdds.scm.storage.BlockExtendedInputStream;
+import org.apache.hadoop.hdds.scm.storage.BlockLocationInfo;
 import org.apache.hadoop.ozone.client.io.BadDataLocationException;
 import org.apache.hadoop.ozone.client.io.ECBlockInputStreamFactory;
 import org.apache.hadoop.ozone.client.io.ECBlockInputStreamProxy;
-import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
@@ -93,7 +93,7 @@
   public void testAvailableDataLocations() {
     Map<DatanodeDetails, Integer> dnMap =
         ECStreamTestUtil.createIndexMap(1, 2, 3, 4, 5);
-    OmKeyLocationInfo blockInfo =
+    BlockLocationInfo blockInfo =
         ECStreamTestUtil.createKeyInfo(repConfig, 1024, dnMap);
     Assert.assertEquals(1, ECBlockInputStreamProxy.availableDataLocations(
         blockInfo.getPipeline(), 1));
@@ -120,7 +120,7 @@
 
     Map<DatanodeDetails, Integer> dnMap =
         ECStreamTestUtil.createIndexMap(1, 2, 3, 4, 5);
-    OmKeyLocationInfo blockInfo =
+    BlockLocationInfo blockInfo =
         ECStreamTestUtil.createKeyInfo(repConfig, blockLength, dnMap);
 
     try (ECBlockInputStreamProxy bis = createBISProxy(repConfig, blockInfo)) {
@@ -135,7 +135,7 @@
 
     Map<DatanodeDetails, Integer> dnMap =
         ECStreamTestUtil.createIndexMap(1, 2, 3, 4, 5);
-    OmKeyLocationInfo blockInfo =
+    BlockLocationInfo blockInfo =
         ECStreamTestUtil.createKeyInfo(repConfig, blockLength, dnMap);
 
     try (ECBlockInputStreamProxy bis = createBISProxy(repConfig, blockInfo)) {
@@ -150,7 +150,7 @@
 
     Map<DatanodeDetails, Integer> dnMap =
         ECStreamTestUtil.createIndexMap(1, 2, 3, 4, 5);
-    OmKeyLocationInfo blockInfo =
+    BlockLocationInfo blockInfo =
         ECStreamTestUtil.createKeyInfo(repConfig, blockLength, dnMap);
 
     dataGenerator = new SplittableRandom(randomSeed);
@@ -172,7 +172,7 @@
 
     Map<DatanodeDetails, Integer> dnMap =
         ECStreamTestUtil.createIndexMap(1, 2, 3, 4, 5);
-    OmKeyLocationInfo blockInfo =
+    BlockLocationInfo blockInfo =
         ECStreamTestUtil.createKeyInfo(repConfig, blockLength, dnMap);
 
     try (ECBlockInputStreamProxy bis = createBISProxy(repConfig, blockInfo)) {
@@ -203,7 +203,7 @@
 
     Map<DatanodeDetails, Integer> dnMap =
         ECStreamTestUtil.createIndexMap(1, 2, 3, 4, 5);
-    OmKeyLocationInfo blockInfo =
+    BlockLocationInfo blockInfo =
         ECStreamTestUtil.createKeyInfo(repConfig, blockLength, dnMap);
 
     ByteBuffer readBuffer = ByteBuffer.allocate(100);
@@ -231,7 +231,7 @@
 
     Map<DatanodeDetails, Integer> dnMap =
         ECStreamTestUtil.createIndexMap(2, 3, 4, 5);
-    OmKeyLocationInfo blockInfo =
+    BlockLocationInfo blockInfo =
         ECStreamTestUtil.createKeyInfo(repConfig, blockLength, dnMap);
 
     ByteBuffer readBuffer = ByteBuffer.allocate(100);
@@ -259,7 +259,7 @@
 
     Map<DatanodeDetails, Integer> dnMap =
         ECStreamTestUtil.createIndexMap(1, 2, 3, 4, 5);
-    OmKeyLocationInfo blockInfo =
+    BlockLocationInfo blockInfo =
         ECStreamTestUtil.createKeyInfo(repConfig, blockLength, dnMap);
 
     ByteBuffer readBuffer = ByteBuffer.allocate(100);
@@ -301,7 +301,7 @@
 
     Map<DatanodeDetails, Integer> dnMap =
         ECStreamTestUtil.createIndexMap(1, 2, 3, 4, 5);
-    OmKeyLocationInfo blockInfo =
+    BlockLocationInfo blockInfo =
         ECStreamTestUtil.createKeyInfo(repConfig, blockLength, dnMap);
 
     ByteBuffer readBuffer = ByteBuffer.allocate(100);
@@ -352,7 +352,7 @@
   }
 
   private ECBlockInputStreamProxy createBISProxy(ECReplicationConfig rConfig,
-      OmKeyLocationInfo blockInfo) {
+      BlockLocationInfo blockInfo) {
     return new ECBlockInputStreamProxy(
         rConfig, blockInfo, true, null, null, streamFactory);
   }
@@ -382,7 +382,7 @@
     @Override
     public BlockExtendedInputStream create(boolean missingLocations,
         List<DatanodeDetails> failedDatanodes,
-        ReplicationConfig repConfig, OmKeyLocationInfo blockInfo,
+        ReplicationConfig repConfig, BlockLocationInfo blockInfo,
         boolean verifyChecksum, XceiverClientFactory xceiverFactory,
         Function<BlockID, Pipeline> refreshFunction) {
       this.failedLocations = failedDatanodes;
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestECBlockReconstructedInputStream.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestECBlockReconstructedInputStream.java
index ee3f50a..823ddd5 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestECBlockReconstructedInputStream.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestECBlockReconstructedInputStream.java
@@ -20,11 +20,11 @@
 import org.apache.hadoop.hdds.client.BlockID;
 import org.apache.hadoop.hdds.client.ECReplicationConfig;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.scm.storage.BlockLocationInfo;
 import org.apache.hadoop.io.ByteBufferPool;
 import org.apache.hadoop.io.ElasticByteBufferPool;
 import org.apache.hadoop.ozone.client.io.ECBlockReconstructedInputStream;
 import org.apache.hadoop.ozone.client.io.ECBlockReconstructedStripeInputStream;
-import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -72,7 +72,7 @@
 
   private ECBlockReconstructedStripeInputStream createStripeInputStream(
       Map<DatanodeDetails, Integer> dnMap, long blockLength) {
-    OmKeyLocationInfo keyInfo =
+    BlockLocationInfo keyInfo =
         ECStreamTestUtil.createKeyInfo(repConfig, blockLength, dnMap);
     streamFactory.setCurrentPipeline(keyInfo.getPipeline());
     return new ECBlockReconstructedStripeInputStream(repConfig, keyInfo, true,
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestECBlockReconstructedStripeInputStream.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestECBlockReconstructedStripeInputStream.java
index 0eff8f9..b111b12 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestECBlockReconstructedStripeInputStream.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestECBlockReconstructedStripeInputStream.java
@@ -19,6 +19,7 @@
 
 import org.apache.hadoop.hdds.client.ECReplicationConfig;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.scm.storage.BlockLocationInfo;
 import org.apache.hadoop.io.ByteBufferPool;
 import org.apache.hadoop.io.ElasticByteBufferPool;
 import org.apache.hadoop.ozone.client.io.ECBlockInputStream;
@@ -26,7 +27,6 @@
 import org.apache.hadoop.ozone.client.io.InsufficientLocationsException;
 import org.apache.hadoop.ozone.client.rpc.read.ECStreamTestUtil.TestBlockInputStreamFactory;
 import org.apache.hadoop.ozone.client.rpc.read.ECStreamTestUtil.TestBlockInputStream;
-import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
 
 import org.junit.After;
 import org.junit.Assert;
@@ -82,7 +82,7 @@
   @Test
   public void testSufficientLocations() throws IOException {
     // One chunk, only 1 location.
-    OmKeyLocationInfo keyInfo = ECStreamTestUtil
+    BlockLocationInfo keyInfo = ECStreamTestUtil
         .createKeyInfo(repConfig, 1, ONEMB);
     try (ECBlockInputStream ecb = createInputStream(keyInfo)) {
       Assert.assertTrue(ecb.hasSufficientLocations());
@@ -168,7 +168,7 @@
       streamFactory = new TestBlockInputStreamFactory();
       addDataStreamsToFactory(dataBufs, parity);
 
-      OmKeyLocationInfo keyInfo = ECStreamTestUtil.createKeyInfo(repConfig,
+      BlockLocationInfo keyInfo = ECStreamTestUtil.createKeyInfo(repConfig,
           stripeSize() * 3 + partialStripeSize, dnMap);
       streamFactory.setCurrentPipeline(keyInfo.getPipeline());
 
@@ -223,7 +223,7 @@
     // from the parity and padded blocks 2 and 3.
     Map<DatanodeDetails, Integer> dnMap =
         ECStreamTestUtil.createIndexMap(4, 5);
-    OmKeyLocationInfo keyInfo =
+    BlockLocationInfo keyInfo =
         ECStreamTestUtil.createKeyInfo(repConfig, blockLength, dnMap);
     streamFactory.setCurrentPipeline(keyInfo.getPipeline());
     dataGen = new SplittableRandom(randomSeed);
@@ -265,7 +265,7 @@
     // from the parity and padded blocks 2 and 3.
     Map<DatanodeDetails, Integer> dnMap =
         ECStreamTestUtil.createIndexMap(4, 5);
-    OmKeyLocationInfo keyInfo =
+    BlockLocationInfo keyInfo =
         ECStreamTestUtil.createKeyInfo(repConfig, blockLength, dnMap);
     streamFactory.setCurrentPipeline(keyInfo.getPipeline());
     dataGen = new SplittableRandom(randomSeed);
@@ -322,7 +322,7 @@
       addDataStreamsToFactory(dataBufs, parity);
       ByteBuffer[] bufs = allocateByteBuffers(repConfig);
 
-      OmKeyLocationInfo keyInfo =
+      BlockLocationInfo keyInfo =
           ECStreamTestUtil.createKeyInfo(repConfig, blockLength, dnMap);
       streamFactory.setCurrentPipeline(keyInfo.getPipeline());
       dataGen = new SplittableRandom(randomSeed);
@@ -365,7 +365,7 @@
     // from the parity and padded blocks 2 and 3.
     Map<DatanodeDetails, Integer> dnMap =
         ECStreamTestUtil.createIndexMap(4, 5);
-    OmKeyLocationInfo keyInfo =
+    BlockLocationInfo keyInfo =
         ECStreamTestUtil.createKeyInfo(repConfig, blockLength, dnMap);
     streamFactory.setCurrentPipeline(keyInfo.getPipeline());
     try (ECBlockReconstructedStripeInputStream ecb =
@@ -407,7 +407,7 @@
       streamFactory = new TestBlockInputStreamFactory();
       addDataStreamsToFactory(dataBufs, parity);
 
-      OmKeyLocationInfo keyInfo = ECStreamTestUtil.createKeyInfo(repConfig,
+      BlockLocationInfo keyInfo = ECStreamTestUtil.createKeyInfo(repConfig,
           stripeSize() * 3 + partialStripeSize, dnMap);
       streamFactory.setCurrentPipeline(keyInfo.getPipeline());
 
@@ -461,7 +461,7 @@
   public void testSeekToPartialOffsetFails() {
     Map<DatanodeDetails, Integer> dnMap =
         ECStreamTestUtil.createIndexMap(1, 4, 5);
-    OmKeyLocationInfo keyInfo = ECStreamTestUtil.createKeyInfo(repConfig,
+    BlockLocationInfo keyInfo = ECStreamTestUtil.createKeyInfo(repConfig,
         stripeSize() * 3, dnMap);
     streamFactory.setCurrentPipeline(keyInfo.getPipeline());
 
@@ -503,7 +503,7 @@
       // Data block index 3 is missing and needs recovered initially.
       Map<DatanodeDetails, Integer> dnMap =
           ECStreamTestUtil.createIndexMap(1, 2, 4, 5);
-      OmKeyLocationInfo keyInfo = ECStreamTestUtil.createKeyInfo(repConfig,
+      BlockLocationInfo keyInfo = ECStreamTestUtil.createKeyInfo(repConfig,
           stripeSize() * 3 + partialStripeSize, dnMap);
       streamFactory.setCurrentPipeline(keyInfo.getPipeline());
 
@@ -583,7 +583,7 @@
     // when containers are reported by SCM.
     Map<DatanodeDetails, Integer> dnMap =
           ECStreamTestUtil.createIndexMap(1, 2, 3, 4, 5);
-    OmKeyLocationInfo keyInfo = ECStreamTestUtil.createKeyInfo(repConfig,
+    BlockLocationInfo keyInfo = ECStreamTestUtil.createKeyInfo(repConfig,
         blockLength, dnMap);
     streamFactory.setCurrentPipeline(keyInfo.getPipeline());
 
@@ -609,7 +609,7 @@
 
     Map<DatanodeDetails, Integer> dnMap =
         ECStreamTestUtil.createIndexMap(1, 2, 3, 4, 5);
-    OmKeyLocationInfo keyInfo = ECStreamTestUtil.createKeyInfo(repConfig,
+    BlockLocationInfo keyInfo = ECStreamTestUtil.createKeyInfo(repConfig,
         stripeSize() * 3 + partialStripeSize, dnMap);
     streamFactory.setCurrentPipeline(keyInfo.getPipeline());
 
@@ -643,7 +643,7 @@
   }
 
   private ECBlockReconstructedStripeInputStream createInputStream(
-      OmKeyLocationInfo keyInfo) {
+      BlockLocationInfo keyInfo) {
     return new ECBlockReconstructedStripeInputStream(repConfig, keyInfo, true,
         null, null, streamFactory, bufferPool, ecReconstructExecutor);
   }
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/freon/TestOmBucketReadWriteFileOps.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/freon/TestOmBucketReadWriteFileOps.java
new file mode 100644
index 0000000..a87b6a6
--- /dev/null
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/freon/TestOmBucketReadWriteFileOps.java
@@ -0,0 +1,326 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.freon;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.ozone.MiniOzoneCluster;
+import org.apache.hadoop.ozone.OzoneConsts;
+import org.apache.hadoop.ozone.client.ObjectStore;
+import org.apache.hadoop.ozone.client.OzoneClientFactory;
+import org.apache.hadoop.ozone.client.OzoneVolume;
+import org.apache.hadoop.ozone.om.OMConfigKeys;
+import org.apache.hadoop.ozone.om.helpers.BucketLayout;
+import org.apache.hadoop.ozone.om.lock.OMLockMetrics;
+import org.apache.ozone.test.GenericTestUtils;
+import org.apache.ratis.server.RaftServer;
+import org.apache.ratis.server.raftlog.RaftLog;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.slf4j.event.Level;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.net.URI;
+
+/**
+ * Test for OmBucketReadWriteFileOps.
+ */
+public class TestOmBucketReadWriteFileOps {
+
+  private String path;
+  private OzoneConfiguration conf = null;
+  private MiniOzoneCluster cluster = null;
+  private ObjectStore store = null;
+  private static final Logger LOG =
+      LoggerFactory.getLogger(TestOmBucketReadWriteFileOps.class);
+
+  @Before
+  public void setup() {
+    path = GenericTestUtils
+        .getTempPath(TestOmBucketReadWriteFileOps.class.getSimpleName());
+    GenericTestUtils.setLogLevel(RaftLog.LOG, Level.DEBUG);
+    GenericTestUtils.setLogLevel(RaftServer.LOG, Level.DEBUG);
+    File baseDir = new File(path);
+    baseDir.mkdirs();
+  }
+
+  /**
+   * Shutdown MiniDFSCluster.
+   */
+  private void shutdown() throws IOException {
+    if (cluster != null) {
+      cluster.shutdown();
+      FileUtils.deleteDirectory(new File(path));
+    }
+  }
+
+  /**
+   * Create a MiniDFSCluster for testing.
+   *
+   * @throws IOException
+   */
+  private void startCluster() throws Exception {
+    conf = getOzoneConfiguration();
+    conf.set(OMConfigKeys.OZONE_DEFAULT_BUCKET_LAYOUT,
+        BucketLayout.LEGACY.name());
+    cluster = MiniOzoneCluster.newBuilder(conf).setNumDatanodes(5).build();
+    cluster.waitForClusterToBeReady();
+    cluster.waitTobeOutOfSafeMode();
+
+    store = OzoneClientFactory.getRpcClient(conf).getObjectStore();
+  }
+
+  protected OzoneConfiguration getOzoneConfiguration() {
+    return new OzoneConfiguration();
+  }
+
+  @Test
+  public void testOmBucketReadWriteFileOps() throws Exception {
+    try {
+      startCluster();
+      FileOutputStream out = FileUtils.openOutputStream(new File(path,
+          "conf"));
+      cluster.getConf().writeXml(out);
+      out.getFD().sync();
+      out.close();
+
+      verifyFreonCommand(new ParameterBuilder().setTotalThreadCount(10)
+          .setNumOfReadOperations(10).setNumOfWriteOperations(5)
+          .setFileCountForRead(10).setFileCountForWrite(5));
+      verifyFreonCommand(
+          new ParameterBuilder().setVolumeName("vol2").setBucketName("bucket1")
+              .setPrefixFilePath("/dir1/dir2/dir3").setTotalThreadCount(10)
+              .setNumOfReadOperations(10).setNumOfWriteOperations(5)
+              .setFileCountForRead(10).setFileCountForWrite(5));
+      verifyFreonCommand(
+          new ParameterBuilder().setVolumeName("vol3").setBucketName("bucket1")
+              .setPrefixFilePath("/").setTotalThreadCount(15)
+              .setNumOfReadOperations(5).setNumOfWriteOperations(3)
+              .setFileCountForRead(5).setFileCountForWrite(3));
+      verifyFreonCommand(
+          new ParameterBuilder().setVolumeName("vol4").setBucketName("bucket1")
+              .setPrefixFilePath("/dir1/").setTotalThreadCount(10)
+              .setNumOfReadOperations(5).setNumOfWriteOperations(3)
+              .setFileCountForRead(5).setFileCountForWrite(3).
+              setFileSizeInBytes(64).setBufferSize(16));
+      verifyFreonCommand(
+          new ParameterBuilder().setVolumeName("vol5").setBucketName("bucket1")
+              .setPrefixFilePath("/dir1/dir2/dir3").setTotalThreadCount(10)
+              .setNumOfReadOperations(5).setNumOfWriteOperations(0)
+              .setFileCountForRead(5));
+      verifyFreonCommand(
+          new ParameterBuilder().setVolumeName("vol6").setBucketName("bucket1")
+              .setPrefixFilePath("/dir1/dir2/dir3/dir4").setTotalThreadCount(20)
+              .setNumOfReadOperations(0).setNumOfWriteOperations(5)
+              .setFileCountForRead(0).setFileCountForWrite(5));
+    } finally {
+      shutdown();
+    }
+  }
+
+  private void verifyFreonCommand(ParameterBuilder parameterBuilder)
+      throws IOException {
+    store.createVolume(parameterBuilder.volumeName);
+    OzoneVolume volume = store.getVolume(parameterBuilder.volumeName);
+    volume.createBucket(parameterBuilder.bucketName);
+    String rootPath = "o3fs://" + parameterBuilder.bucketName + "." +
+            parameterBuilder.volumeName + parameterBuilder.prefixFilePath;
+    String confPath = new File(path, "conf").getAbsolutePath();
+    new Freon().execute(
+        new String[]{"-conf", confPath, "obrwf", "-P", rootPath,
+            "-r", String.valueOf(parameterBuilder.fileCountForRead),
+            "-w", String.valueOf(parameterBuilder.fileCountForWrite),
+            "-g", String.valueOf(parameterBuilder.fileSizeInBytes),
+            "-b", String.valueOf(parameterBuilder.bufferSize),
+            "-l", String.valueOf(parameterBuilder.length),
+            "-c", String.valueOf(parameterBuilder.totalThreadCount),
+            "-T", String.valueOf(parameterBuilder.readThreadPercentage),
+            "-R", String.valueOf(parameterBuilder.numOfReadOperations),
+            "-W", String.valueOf(parameterBuilder.numOfWriteOperations),
+            "-n", String.valueOf(1)});
+
+    LOG.info("Started verifying OM bucket read/write ops file generation...");
+    FileSystem fileSystem = FileSystem.get(URI.create(rootPath),
+        conf);
+    Path rootDir = new Path(rootPath.concat(OzoneConsts.OM_KEY_PREFIX));
+    FileStatus[] fileStatuses = fileSystem.listStatus(rootDir);
+    verifyFileCreation(2, fileStatuses, true);
+
+    Path readDir = new Path(rootPath.concat("/readPath"));
+    FileStatus[] readFileStatuses = fileSystem.listStatus(readDir);
+    verifyFileCreation(parameterBuilder.fileCountForRead, readFileStatuses,
+        false);
+
+    int readThreadCount = (parameterBuilder.readThreadPercentage *
+        parameterBuilder.totalThreadCount) / 100;
+    int writeThreadCount = parameterBuilder.totalThreadCount - readThreadCount;
+
+    Path writeDir = new Path(rootPath.concat("/writePath"));
+    FileStatus[] writeFileStatuses = fileSystem.listStatus(writeDir);
+    verifyFileCreation(writeThreadCount * parameterBuilder.fileCountForWrite *
+        parameterBuilder.numOfWriteOperations, writeFileStatuses, false);
+
+    verifyOMLockMetrics(cluster.getOzoneManager().getMetadataManager().getLock()
+        .getOMLockMetrics());
+  }
+
+  private void verifyFileCreation(int expectedCount, FileStatus[] fileStatuses,
+                          boolean checkDirectoryCount) {
+    int actual = 0;
+    if (checkDirectoryCount) {
+      for (FileStatus fileStatus : fileStatuses) {
+        if (fileStatus.isDirectory()) {
+          ++actual;
+        }
+      }
+    } else {
+      for (FileStatus fileStatus : fileStatuses) {
+        if (fileStatus.isFile()) {
+          ++actual;
+        }
+      }
+    }
+    Assert.assertEquals("Mismatch Count!", expectedCount, actual);
+  }
+
+  private void verifyOMLockMetrics(OMLockMetrics omLockMetrics) {
+    String readLockWaitingTimeMsStat =
+        omLockMetrics.getReadLockWaitingTimeMsStat();
+    LOG.info("Read Lock Waiting Time Stat: " + readLockWaitingTimeMsStat);
+    LOG.info("Longest Read Lock Waiting Time (ms): " +
+        omLockMetrics.getLongestReadLockWaitingTimeMs());
+    int readWaitingSamples =
+        Integer.parseInt(readLockWaitingTimeMsStat.split(" ")[2]);
+    Assert.assertTrue("Read Lock Waiting Samples should be positive",
+        readWaitingSamples > 0);
+
+    String readLockHeldTimeMsStat = omLockMetrics.getReadLockHeldTimeMsStat();
+    LOG.info("Read Lock Held Time Stat: " + readLockHeldTimeMsStat);
+    LOG.info("Longest Read Lock Held Time (ms): " +
+        omLockMetrics.getLongestReadLockHeldTimeMs());
+    int readHeldSamples =
+        Integer.parseInt(readLockHeldTimeMsStat.split(" ")[2]);
+    Assert.assertTrue("Read Lock Held Samples should be positive",
+        readHeldSamples > 0);
+
+    String writeLockWaitingTimeMsStat =
+        omLockMetrics.getWriteLockWaitingTimeMsStat();
+    LOG.info("Write Lock Waiting Time Stat: " + writeLockWaitingTimeMsStat);
+    LOG.info("Longest Write Lock Waiting Time (ms): " +
+        omLockMetrics.getLongestWriteLockWaitingTimeMs());
+    int writeWaitingSamples =
+        Integer.parseInt(writeLockWaitingTimeMsStat.split(" ")[2]);
+    Assert.assertTrue("Write Lock Waiting Samples should be positive",
+        writeWaitingSamples > 0);
+
+    String writeLockHeldTimeMsStat = omLockMetrics.getWriteLockHeldTimeMsStat();
+    LOG.info("Write Lock Held Time Stat: " + writeLockHeldTimeMsStat);
+    LOG.info("Longest Write Lock Held Time (ms): " +
+        omLockMetrics.getLongestWriteLockHeldTimeMs());
+    int writeHeldSamples =
+        Integer.parseInt(writeLockHeldTimeMsStat.split(" ")[2]);
+    Assert.assertTrue("Write Lock Held Samples should be positive",
+        writeHeldSamples > 0);
+  }
+
+  private static class ParameterBuilder {
+
+    private String volumeName = "vol1";
+    private String bucketName = "bucket1";
+    private String prefixFilePath = "/dir1/dir2";
+    private int fileCountForRead = 100;
+    private int fileCountForWrite = 10;
+    private long fileSizeInBytes = 256;
+    private int bufferSize = 64;
+    private int length = 10;
+    private int totalThreadCount = 100;
+    private int readThreadPercentage = 90;
+    private int numOfReadOperations = 50;
+    private int numOfWriteOperations = 10;
+
+    private ParameterBuilder setVolumeName(String volumeNameParam) {
+      volumeName = volumeNameParam;
+      return this;
+    }
+
+    private ParameterBuilder setBucketName(String bucketNameParam) {
+      bucketName = bucketNameParam;
+      return this;
+    }
+
+    private ParameterBuilder setPrefixFilePath(String prefixFilePathParam) {
+      prefixFilePath = prefixFilePathParam;
+      return this;
+    }
+
+    private ParameterBuilder setFileCountForRead(int fileCountForReadParam) {
+      fileCountForRead = fileCountForReadParam;
+      return this;
+    }
+
+    private ParameterBuilder setFileCountForWrite(int fileCountForWriteParam) {
+      fileCountForWrite = fileCountForWriteParam;
+      return this;
+    }
+
+    private ParameterBuilder setFileSizeInBytes(long fileSizeInBytesParam) {
+      fileSizeInBytes = fileSizeInBytesParam;
+      return this;
+    }
+
+    private ParameterBuilder setBufferSize(int bufferSizeParam) {
+      bufferSize = bufferSizeParam;
+      return this;
+    }
+
+    private ParameterBuilder setLength(int lengthParam) {
+      length = lengthParam;
+      return this;
+    }
+
+    private ParameterBuilder setTotalThreadCount(int totalThreadCountParam) {
+      totalThreadCount = totalThreadCountParam;
+      return this;
+    }
+
+    private ParameterBuilder setReadThreadPercentage(
+        int readThreadPercentageParam) {
+      readThreadPercentage = readThreadPercentageParam;
+      return this;
+    }
+
+    private ParameterBuilder setNumOfReadOperations(
+        int numOfReadOperationsParam) {
+      numOfReadOperations = numOfReadOperationsParam;
+      return this;
+    }
+
+    private ParameterBuilder setNumOfWriteOperations(
+        int numOfWriteOperationsParam) {
+      numOfWriteOperations = numOfWriteOperationsParam;
+      return this;
+    }
+  }
+}
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHAMetadataOnly.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHAMetadataOnly.java
index 0bb5011..e1874c3 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHAMetadataOnly.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHAMetadataOnly.java
@@ -40,6 +40,7 @@
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.ozone.test.GenericTestUtils;
 import org.apache.log4j.Logger;
+import org.apache.ozone.test.tag.Flaky;
 import org.apache.ratis.protocol.ClientId;
 import org.apache.ratis.protocol.Message;
 import org.apache.ratis.protocol.RaftClientReply;
@@ -281,6 +282,7 @@
   }
 
   @Test
+  @Flaky("HDDS-6644")
   public void testReadRequest() throws Exception {
     String volumeName = "volume" + RandomStringUtils.randomNumeric(5);
     ObjectStore objectStore = getObjectStore();
@@ -318,6 +320,7 @@
   }
 
   @Test
+  @Flaky("HDDS-6642")
   public void testListVolumes() throws Exception {
     String userName = UserGroupInformation.getCurrentUser().getUserName();
     ObjectStore objectStore = getObjectStore();
diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/security/acl/TestParentAcl.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/security/acl/TestParentAcl.java
index 8862b0f..82dd430 100644
--- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/security/acl/TestParentAcl.java
+++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/security/acl/TestParentAcl.java
@@ -45,6 +45,7 @@
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.authentication.client.AuthenticationException;
 import org.apache.ozone.test.GenericTestUtils;
+import org.apache.ozone.test.tag.Flaky;
 import org.junit.Assert;
 import org.junit.BeforeClass;
 import org.junit.Test;
@@ -129,6 +130,7 @@
   // LIST        LIST         READ      (V1 LIST=>READ)
   // READ_ACL    READ_ACL     READ      (V1 READ_ACL=>READ)
   @Test
+  @Flaky("HDDS-6335")
   public void testKeyAcl()
       throws IOException {
     OzoneObj keyObj;
diff --git a/hadoop-ozone/ozonefs-shaded/pom.xml b/hadoop-ozone/ozonefs-shaded/pom.xml
index 46419e1..9d26a77 100644
--- a/hadoop-ozone/ozonefs-shaded/pom.xml
+++ b/hadoop-ozone/ozonefs-shaded/pom.xml
@@ -53,6 +53,10 @@
           <artifactId>hadoop-annotations</artifactId>
         </exclusion>
         <exclusion>
+          <groupId>org.apache.hadoop.thirdparty</groupId>
+          <artifactId>*</artifactId>
+        </exclusion>
+        <exclusion>
           <groupId>log4j</groupId>
           <artifactId>log4j</artifactId>
         </exclusion>
diff --git a/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/freon/Freon.java b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/freon/Freon.java
index c5b9a3a..914b709 100644
--- a/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/freon/Freon.java
+++ b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/freon/Freon.java
@@ -63,7 +63,8 @@
         GeneratorDatanode.class,
         ClosedContainerReplicator.class,
         StreamingGenerator.class,
-        SCMThroughputBenchmark.class},
+        SCMThroughputBenchmark.class,
+        OmBucketReadWriteFileOps.class},
     versionProvider = HddsVersionProvider.class,
     mixinStandardHelpOptions = true)
 public class Freon extends GenericCli {
diff --git a/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/freon/OmBucketReadWriteFileOps.java b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/freon/OmBucketReadWriteFileOps.java
new file mode 100644
index 0000000..401e71b1
--- /dev/null
+++ b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/freon/OmBucketReadWriteFileOps.java
@@ -0,0 +1,278 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.freon;
+
+import com.codahale.metrics.Timer;
+import org.apache.commons.lang3.RandomStringUtils;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdds.cli.HddsVersionProvider;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.ozone.OzoneConsts;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import picocli.CommandLine.Command;
+import picocli.CommandLine.Option;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.CompletionService;
+import java.util.concurrent.ExecutorCompletionService;
+
+/**
+ * Synthetic read/write file operations workload generator tool.
+ */
+@Command(name = "obrwf",
+    aliases = "om-bucket-read-write-file-ops",
+    description = "Creates files, performs respective read/write " +
+        "operations to measure lock performance.",
+    versionProvider = HddsVersionProvider.class,
+    mixinStandardHelpOptions = true,
+    showDefaultValues = true)
+
+public class OmBucketReadWriteFileOps extends BaseFreonGenerator
+    implements Callable<Void> {
+
+  private static final Logger LOG =
+      LoggerFactory.getLogger(OmBucketReadWriteFileOps.class);
+
+  @Option(names = {"-P", "--root-path"},
+      description = "Root path",
+      defaultValue = "o3fs://bucket1.vol1/dir1/dir2")
+  private String rootPath;
+
+  @Option(names = {"-r", "--file-count-for-read"},
+      description = "Number of files to be written in the read directory.",
+      defaultValue = "100")
+  private int fileCountForRead;
+
+  @Option(names = {"-w", "--file-count-for-write"},
+      description = "Number of files to be written in the write directory.",
+      defaultValue = "10")
+  private int fileCountForWrite;
+
+  @Option(names = {"-g", "--file-size"},
+      description = "Generated data size (in bytes) of each file to be " +
+          "written in each directory.",
+      defaultValue = "256")
+  private long fileSizeInBytes;
+
+  @Option(names = {"-b", "--buffer"},
+      description = "Size of buffer used to generated the file content.",
+      defaultValue = "64")
+  private int bufferSize;
+
+  @Option(names = {"-l", "--name-len"},
+      description = "Length of the random name of directory you want to " +
+          "create.",
+      defaultValue = "10")
+  private int length;
+
+  @Option(names = {"-c", "--total-thread-count"},
+      description = "Total number of threads to be executed.",
+      defaultValue = "100")
+  private int totalThreadCount;
+
+  @Option(names = {"-T", "--read-thread-percentage"},
+      description = "Percentage of the total number of threads to be " +
+          "allocated for read operations. The remaining percentage of " +
+          "threads will be allocated for write operations.",
+      defaultValue = "90")
+  private int readThreadPercentage;
+
+  @Option(names = {"-R", "--num-of-read-operations"},
+      description = "Number of read operations to be performed by each thread.",
+      defaultValue = "50")
+  private int numOfReadOperations;
+
+  @Option(names = {"-W", "--num-of-write-operations"},
+      description = "Number of write operations to be performed by each " +
+          "thread.",
+      defaultValue = "10")
+  private int numOfWriteOperations;
+
+  private Timer timer;
+
+  private ContentGenerator contentGenerator;
+
+  private FileSystem fileSystem;
+
+  private int readThreadCount;
+  private int writeThreadCount;
+
+  @Override
+  public Void call() throws Exception {
+    init();
+
+    readThreadCount = (readThreadPercentage * totalThreadCount) / 100;
+    writeThreadCount = totalThreadCount - readThreadCount;
+
+    print("rootPath: " + rootPath);
+    print("fileCountForRead: " + fileCountForRead);
+    print("fileCountForWrite: " + fileCountForWrite);
+    print("fileSizeInBytes: " + fileSizeInBytes);
+    print("bufferSize: " + bufferSize);
+    print("totalThreadCount: " + totalThreadCount);
+    print("readThreadPercentage: " + readThreadPercentage);
+    print("writeThreadPercentage: " + (100 - readThreadPercentage));
+    print("readThreadCount: " + readThreadCount);
+    print("writeThreadCount: " + writeThreadCount);
+    print("numOfReadOperations: " + numOfReadOperations);
+    print("numOfWriteOperations: " + numOfWriteOperations);
+
+    OzoneConfiguration configuration = createOzoneConfiguration();
+    fileSystem = FileSystem.get(URI.create(rootPath), configuration);
+
+    contentGenerator = new ContentGenerator(fileSizeInBytes, bufferSize);
+    timer = getMetrics().timer("file-create");
+
+    runTests(this::mainMethod);
+    return null;
+  }
+
+  private void mainMethod(long counter) throws Exception {
+
+    int readResult = readOperations();
+    int writeResult = writeOperations();
+
+    print("Total Files Read: " + readResult);
+    print("Total Files Written: " + writeResult * fileCountForWrite);
+
+    // TODO: print read/write lock metrics (HDDS-6435, HDDS-6436).
+  }
+
+  private int readOperations() throws Exception {
+
+    // Create fileCountForRead (defaultValue = 1000) files under
+    // rootPath/readPath directory
+    String readPath =
+        rootPath.concat(OzoneConsts.OM_KEY_PREFIX).concat("readPath");
+    fileSystem.mkdirs(new Path(readPath));
+    createFiles(readPath, fileCountForRead);
+
+    // Start readThreadCount (defaultValue = 90) concurrent read threads
+    // performing numOfReadOperations (defaultValue = 50) iterations
+    // of read operations (fileSystem.listStatus(rootPath/readPath))
+    ExecutorService readService = Executors.newFixedThreadPool(readThreadCount);
+    CompletionService<Integer> readExecutorCompletionService =
+        new ExecutorCompletionService<>(readService);
+    List<Future<Integer>> readFutures = new ArrayList<>();
+    for (int i = 0; i < readThreadCount; i++) {
+      readFutures.add(readExecutorCompletionService.submit(() -> {
+        int readCount = 0;
+        try {
+          for (int j = 0; j < numOfReadOperations; j++) {
+            FileStatus[] status =
+                fileSystem.listStatus(new Path(readPath));
+            readCount += status.length;
+          }
+        } catch (IOException e) {
+          LOG.warn("Exception while listing status ", e);
+        }
+        return readCount;
+      }));
+    }
+
+    int readResult = 0;
+    for (int i = 0; i < readFutures.size(); i++) {
+      try {
+        readResult += readExecutorCompletionService.take().get();
+      } catch (InterruptedException e) {
+        e.printStackTrace();
+      } catch (ExecutionException e) {
+        e.printStackTrace();
+      }
+    }
+    readService.shutdown();
+
+    return readResult;
+  }
+
+  private int writeOperations() throws Exception {
+
+    // Start writeThreadCount (defaultValue = 10) concurrent write threads
+    // performing numOfWriteOperations (defaultValue = 10) iterations
+    // of write operations (createFiles(rootPath/writePath))
+    String writePath =
+        rootPath.concat(OzoneConsts.OM_KEY_PREFIX).concat("writePath");
+    fileSystem.mkdirs(new Path(writePath));
+
+    ExecutorService writeService =
+        Executors.newFixedThreadPool(writeThreadCount);
+    CompletionService<Integer> writeExecutorCompletionService =
+        new ExecutorCompletionService<>(writeService);
+    List<Future<Integer>> writeFutures = new ArrayList<>();
+    for (int i = 0; i < writeThreadCount; i++) {
+      writeFutures.add(writeExecutorCompletionService.submit(() -> {
+        int writeCount = 0;
+        try {
+          for (int j = 0; j < numOfWriteOperations; j++) {
+            createFiles(writePath, fileCountForWrite);
+            writeCount++;
+          }
+        } catch (IOException e) {
+          LOG.warn("Exception while creating file ", e);
+        }
+        return writeCount;
+      }));
+    }
+
+    int writeResult = 0;
+    for (int i = 0; i < writeFutures.size(); i++) {
+      try {
+        writeResult += writeExecutorCompletionService.take().get();
+      } catch (InterruptedException e) {
+        e.printStackTrace();
+      } catch (ExecutionException e) {
+        e.printStackTrace();
+      }
+    }
+    writeService.shutdown();
+
+    return writeResult;
+  }
+
+  private void createFile(String dir, long counter) throws Exception {
+    String fileName = dir.concat(OzoneConsts.OM_KEY_PREFIX)
+        .concat(RandomStringUtils.randomAlphanumeric(length));
+    Path file = new Path(fileName);
+    if (LOG.isDebugEnabled()) {
+      LOG.debug("FilePath:{}", file);
+    }
+    timer.time(() -> {
+      try (FSDataOutputStream output = fileSystem.create(file)) {
+        contentGenerator.write(output);
+      }
+      return null;
+    });
+  }
+
+  private void createFiles(String dir, int fileCount) throws Exception {
+    for (int i = 0; i < fileCount; i++) {
+      createFile(dir, i);
+    }
+  }
+}