diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
index 51fc564a97584..77355be0c2664 100755
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
@@ -836,6 +836,14 @@ public class DFSConfigKeys extends CommonConfigurationKeys {
   public static final int     DFS_DATANODE_DIRECTORYSCAN_INTERVAL_DEFAULT = 21600;
   public static final String  DFS_DATANODE_DIRECTORYSCAN_THREADS_KEY = "dfs.datanode.directoryscan.threads";
   public static final int     DFS_DATANODE_DIRECTORYSCAN_THREADS_DEFAULT = 1;
+  public static final String  DFS_DATANODE_RECONCILE_BLOCKS_BATCH_SIZE =
+      "dfs.datanode.reconcile.blocks.batch.size";
+  public static final int
+      DFS_DATANODE_RECONCILE_BLOCKS_BATCH_SIZE_DEFAULT = 1000;
+  public static final String DFS_DATANODE_RECONCILE_BLOCKS_BATCH_INTERVAL
+      = "dfs.datanode.reconcile.blocks.batch.interval";
+  public static final long
+      DFS_DATANODE_RECONCILE_BLOCKS_BATCH_INTERVAL_DEFAULT = 2000;
 
   public static final String DFS_DATANODE_DISK_CHECK_MIN_GAP_KEY =
       "dfs.datanode.disk.check.min.gap";
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DirectoryScanner.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DirectoryScanner.java
index a3bceec9a96a7..7a8569d39008e 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DirectoryScanner.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DirectoryScanner.java
@@ -65,7 +65,8 @@ public class DirectoryScanner implements Runnable {
       LoggerFactory.getLogger(DirectoryScanner.class);
 
   private static final int DEFAULT_MAP_SIZE = 32768;
-  private static final int RECONCILE_BLOCKS_BATCH_SIZE = 1000;
+  private final int reconcileBlocksBatchSize;
+  private final long reconcileBlocksBatchInterval;
   private final FsDatasetSpi<?> dataset;
   private final ExecutorService reportCompileThreadPool;
   private final ScheduledExecutorService masterThread;
@@ -315,6 +316,41 @@ public DirectoryScanner(FsDatasetSpi<?> dataset, Configuration conf) {
 
     masterThread =
         new ScheduledThreadPoolExecutor(1, new Daemon.DaemonFactory());
+
+    int reconcileBatchSize =
+        conf.getInt(DFSConfigKeys.
+                DFS_DATANODE_RECONCILE_BLOCKS_BATCH_SIZE,
+            DFSConfigKeys.
+                DFS_DATANODE_RECONCILE_BLOCKS_BATCH_SIZE_DEFAULT);
+
+    if (reconcileBatchSize <= 0) {
+      LOG.warn("Invalid value configured for " +
+              "dfs.datanode.reconcile.blocks.batch.size, " +
+              "should be greater than 0, Using default.");
+      reconcileBatchSize =
+          DFSConfigKeys.
+              DFS_DATANODE_RECONCILE_BLOCKS_BATCH_SIZE_DEFAULT;
+    }
+
+    reconcileBlocksBatchSize = reconcileBatchSize;
+
+    long reconcileBatchInterval =
+        conf.getTimeDuration(DFSConfigKeys.
+                DFS_DATANODE_RECONCILE_BLOCKS_BATCH_INTERVAL,
+            DFSConfigKeys.
+                DFS_DATANODE_RECONCILE_BLOCKS_BATCH_INTERVAL_DEFAULT,
+            TimeUnit.MILLISECONDS);
+
+    if (reconcileBatchInterval <= 0) {
+      LOG.warn("Invalid value configured for " +
+              "dfs.datanode.reconcile.blocks.batch.interval, " +
+              "should be greater than 0, Using default.");
+      reconcileBatchInterval =
+          DFSConfigKeys.
+              DFS_DATANODE_RECONCILE_BLOCKS_BATCH_INTERVAL_DEFAULT;
+    }
+
+    reconcileBlocksBatchInterval = reconcileBatchInterval;
   }
 
   /**
@@ -428,16 +464,16 @@ public void reconcile() throws IOException {
     LOG.debug("reconcile start DirectoryScanning");
     scan();
 
-    // HDFS-14476: run checkAndUpadte with batch to avoid holding the lock too
+    // HDFS-14476: run checkAndUpdate with batch to avoid holding the lock too
     // long
     int loopCount = 0;
     synchronized (diffs) {
       for (final Map.Entry<String, ScanInfo> entry : diffs.getEntries()) {
         dataset.checkAndUpdate(entry.getKey(), entry.getValue());
 
-        if (loopCount % RECONCILE_BLOCKS_BATCH_SIZE == 0) {
+        if (loopCount % reconcileBlocksBatchSize == 0) {
           try {
-            Thread.sleep(2000);
+            Thread.sleep(reconcileBlocksBatchInterval);
           } catch (InterruptedException e) {
             // do nothing
           }
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml b/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml
index a30889bdd0042..6e8e36e1cf0dc 100755
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml
@@ -873,6 +873,18 @@
   </description>
 </property>
 
+<property>
+  <name>dfs.datanode.reconcile.blocks.batch.size</name>
+  <value>1000</value>
+  <description>Setting this to define reconcile batch size.</description>
+</property>
+
+<property>
+  <name>dfs.datanode.reconcile.blocks.batch.interval</name>
+  <value>2000</value>
+  <description>Setting this to define interval between batches.</description>
+</property>
+
 <property>
   <name>dfs.heartbeat.interval</name>
   <value>3s</value>