HDFS-17216. Distcp: When handle the small files, the bandwidth parameter will be invalid, fix this bug.

Author: xiaojunxiang2023

hadoop-tools/hadoop-distcp/pom.xml

@@ -114,6 +114,12 @@
       <artifactId>assertj-core</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.hamcrest</groupId>
+      <artifactId>hamcrest-library</artifactId>
+      <version>1.3</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>

hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/util/ThrottledInputStream.java

@@ -120,12 +120,11 @@ public long getTotalBytesRead() {
    * @return Read rate, in bytes/sec.
    */
   public long getBytesPerSec() {
-    long elapsed = (System.currentTimeMillis() - startTime) / 1000;
-    if (elapsed == 0) {
-      return bytesRead;
-    } else {
-      return bytesRead / elapsed;
+    if (bytesRead == 0){
+      return 0;
     }
+    float elapsed = (System.currentTimeMillis() - startTime) / 1000.0f;
+    return (long) (bytesRead / elapsed);
   }
 
   /**

hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/util/TestThrottledInputStream.java

@@ -22,6 +22,8 @@
 import org.slf4j.LoggerFactory;
 import org.apache.hadoop.io.IOUtils;
 import org.junit.Assert;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.junit.Assert.assertThat;
 import org.junit.Test;
 
 import java.io.*;
@@ -67,6 +69,40 @@ public void testRead() {
     }
   }
 
+  @Test
+  public void testThrottleRead() {
+    int testFileCnt = 100;
+    int fileSize = 19;
+    int bandwidth= 20;
+    File[] srcFiles = new File[testFileCnt];
+    File destFile;
+    try {
+      destFile = createFile(testFileCnt * 100 * 1024);
+      destFile.deleteOnExit();
+      // create file
+      for (int i = 0; i < srcFiles.length; i++) {
+        srcFiles[i] = createFile(fileSize * 1024);
+        srcFiles[i].deleteOnExit();
+      }
+
+      // copy srcFiles
+      long begin = System.currentTimeMillis();
+      LOG.info("begin: " + begin);
+
+      for (File srcFile : srcFiles) {
+        LOG.info("fileLength: " + srcFiles.length);
+        copyAndAssertWhenFloatCalc(srcFile, destFile, bandwidth * 1024 * 1024);
+      }
+
+      long end = System.currentTimeMillis();
+      LOG.info("end: " + end);
+      assertThat((int) (end - begin) / 1000,
+              greaterThanOrEqualTo(testFileCnt * fileSize / bandwidth));
+    } catch (IOException e) {
+      LOG.error("Exception encountered ", e);
+    }
+  }
+
   private long copyAndAssert(File tmpFile, File outFile,
                              long maxBandwidth, float factor,
                              int sleepTime, CB flag) throws IOException {
@@ -101,6 +137,23 @@ private long copyAndAssert(File tmpFile, File outFile,
     return bandwidth;
   }
 
+  private void copyAndAssertWhenFloatCalc(File tmpFile, File outFile, long maxBPS)
+          throws IOException {
+    ThrottledInputStream in = new ThrottledInputStream(new FileInputStream(tmpFile), maxBPS);
+    OutputStream out = new FileOutputStream(outFile);
+    try {
+      copyBytes(in, out, BUFF_SIZE);
+      LOG.info("{}", in);
+      Assert.assertEquals(in.getTotalBytesRead(), tmpFile.length());
+
+      long bytesPerSec = in.getBytesPerSec();
+      Assert.assertTrue( bytesPerSec < maxBPS);
+    } finally {
+      IOUtils.closeStream(in);
+      IOUtils.closeStream(out);
+    }
+  }
+
   private static void copyBytesWithOffset(InputStream in, OutputStream out, int buffSize)
     throws IOException {