HDFS-7681. Change ReplicaInputStreams constructor to take InputStream(s) instead of FileDescriptor(s). Contributed by Joe Pallas

Conflicts:
	hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt

hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt

@@ -288,6 +288,9 @@ Release 2.7.0 - UNRELEASED
     type instead of concrete classes Block and ReplicaInfo.  (David Powell
     and Joe Pallas via szetszwo)
 
+    HDFS-7681. Change ReplicaInputStreams constructor to take InputStream(s)
+    instead of FileDescriptor(s).  (Joe Pallas via szetszwo)
+
   OPTIMIZATIONS
 
     HDFS-7454. Reduce memory footprint for AclEntries in NameNode.

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

@@ -33,11 +33,11 @@ public class ReplicaInputStreams implements Closeable {
   private final FsVolumeReference volumeRef;
 
   /** Create an object with a data input stream and a checksum input stream. */
-  public ReplicaInputStreams(FileDescriptor dataFd, FileDescriptor checksumFd,
+  public ReplicaInputStreams(InputStream dataStream, InputStream checksumStream,
       FsVolumeReference volumeRef) {
     this.volumeRef = volumeRef;
-    this.dataIn = new FileInputStream(dataFd);
-    this.checksumIn = new FileInputStream(checksumFd);
+    this.dataIn = dataStream;
+    this.checksumIn = checksumStream;
   }
 
   /** @return the data input stream. */

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsDatasetImpl.java

@@ -644,8 +644,14 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
       if (ckoff > 0) {
         metaInFile.seek(ckoff);
       }
-      return new ReplicaInputStreams(
-          blockInFile.getFD(), metaInFile.getFD(), ref);
+      InputStream blockInStream = new FileInputStream(blockInFile.getFD());
+      try {
+        InputStream metaInStream = new FileInputStream(metaInFile.getFD());
+        return new ReplicaInputStreams(blockInStream, metaInStream, ref);
+      } catch (IOException e) {
+        IOUtils.cleanup(null, blockInStream);
+        throw e;
+      }
     } catch (IOException e) {
       IOUtils.cleanup(null, ref);
       throw e;