Make DataTransfer implements Callable rather than Runnable

Author: LiuGuH

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java

@@ -146,6 +146,7 @@
 import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import javax.annotation.Nullable;
@@ -2895,7 +2896,7 @@ void transferBlock(ExtendedBlock block, DatanodeInfo[] xferTargets,
           xferTargetStorageTypes, xferTargetStorageIDs, block,
           BlockConstructionStage.PIPELINE_SETUP_CREATE, "");
 
-      this.xferService.execute(dataTransferTask);
+      this.xferService.submit(dataTransferTask);
     }
   }
 
@@ -3006,7 +3007,7 @@ CHECKSUM_SIZE depends on CHECKSUM_TYPE (usually, 4 for CRC32)
    * Used for transferring a block of data.  This class
    * sends a piece of data to another DataNode.
    */
-  private class DataTransfer implements Runnable {
+  private class DataTransfer implements Callable<Void> {
     final DatanodeInfo[] targets;
     final StorageType[] targetStorageTypes;
     final private String[] targetStorageIds;
@@ -3064,9 +3065,11 @@ private class DataTransfer implements Runnable {
 
     /**
      * Do the deed, write the bytes
+     *
+     * @return
      */
     @Override
-    public void run() {
+    public Void call() throws IOException {
       incrementXmitsInProgress();
       Socket sock = null;
       DataOutputStream out = null;
@@ -3146,15 +3149,15 @@ public void run() {
         }
       } catch (IOException ie) {
         if (copyBlockCrossNamespace) {
-          throw new RuntimeException(ie);
+          throw ie;
         }
         handleBadBlock(source, ie, false);
         LOG.warn("{}:Failed to transfer {} to {}  got",
             bpReg, source, targets[0], ie);
       } catch (Throwable t) {
         LOG.error("Failed to transfer block {}", source, t);
         if (copyBlockCrossNamespace) {
-          throw new RuntimeException(t);
+          throw t;
         }
       } finally {
         decrementXmitsInProgress();
@@ -3163,6 +3166,7 @@ public void run() {
         IOUtils.closeStream(in);
         IOUtils.closeSocket(sock);
       }
+      return null;
     }
 
     @Override
@@ -4444,7 +4448,7 @@ public void copyBlockCrossNamespace(ExtendedBlock sourceBlk, ExtendedBlock targe
     }
     try {
       result.get(getDnConf().getCopyBlockCrossNamespaceSocketTimeout(), TimeUnit.MILLISECONDS);
-    } catch (Exception e) {
+    } catch (ExecutionException | InterruptedException | TimeoutException e) {
       LOG.error(e.getMessage());
       throw new IOException(e);
     }

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDataTransferProtocol.java

@@ -666,7 +666,7 @@ public void testReleaseVolumeRefIfExceptionThrown()
     }
   }
 
-  @Test(timeout = 90000)
+  @Test
   public void testCopyBlockCrossNamespace()
       throws IOException, InterruptedException, TimeoutException {
     Configuration conf = new HdfsConfiguration();