diff --git a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/CopyCommitter.java b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/CopyCommitter.java
index 38106fae9b9..b86f5ea4149 100644
--- a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/CopyCommitter.java
+++ b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/CopyCommitter.java
@@ -73,6 +73,7 @@ public class CopyCommitter extends FileOutputCommitter {
   private boolean overwrite = false;
   private boolean targetPathExists = true;
   private boolean ignoreFailures = false;
+  private int blocksPerChunk = 0;
 
   /**
    * Create a output committer
@@ -83,6 +84,9 @@ public class CopyCommitter extends FileOutputCommitter {
    */
   public CopyCommitter(Path outputPath, TaskAttemptContext context) throws IOException {
     super(outputPath, context);
+    blocksPerChunk = context.getConfiguration().getInt(
+        DistCpOptionSwitch.BLOCKS_PER_CHUNK.getConfigLabel(), 0);
+    LOG.debug("blocks per chunk {}", blocksPerChunk);
     this.taskAttemptContext = context;
   }
 
@@ -97,7 +101,9 @@ public class CopyCommitter extends FileOutputCommitter {
     ignoreFailures = conf.getBoolean(
         DistCpOptionSwitch.IGNORE_FAILURES.getConfigLabel(), false);
 
-    concatFileChunks(conf);
+    if (blocksPerChunk > 0) {
+      concatFileChunks(conf);
+    }
 
     super.commitJob(jobContext);
 
diff --git a/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/mapred/TestCopyCommitter.java b/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/mapred/TestCopyCommitter.java
index 8d9f8676c2c..912205f24b0 100644
--- a/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/mapred/TestCopyCommitter.java
+++ b/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/mapred/TestCopyCommitter.java
@@ -85,10 +85,6 @@ public class TestCopyCommitter {
   @Before
   public void createMetaFolder() throws IOException {
     config.set(DistCpConstants.CONF_LABEL_META_FOLDER, "/meta");
-    // Unset listing file path since the config is shared by
-    // multiple tests, and some test doesn't set it, such as
-    // testNoCommitAction, but the distcp code will check it.
-    config.set(DistCpConstants.CONF_LABEL_LISTING_FILE_PATH, "");
     Path meta = new Path("/meta");
     cluster.getFileSystem().mkdirs(meta);
   }