[CELEBORN-2047] Reuse FileChannel/FSDataInputStream in PartitionDataReader

### What changes were proposed in this pull request?

Reuse FileChannel/FSDataInputStream in PartitionDataReader to avoid  open too many files.

### Why are the changes needed?

In previous implementations, different PartitionDataReader reused the same FileChannel/FSDataInputStream,but in #3349
changed this logic, so we need to maintain logical consistency

### Does this PR introduce any user-facing change?

No.

### How was this patch tested?

Manual test.

Closes #3445 from Alibaba-HZY/mirror-fix.

Authored-by: daowu.hzy <daowu.hzy@alibaba-inc.com>
Signed-off-by: mingji <fengmingxiao.fmx@alibaba-inc.com>

worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/DfsPartitionDataReader.java

@@ -35,7 +35,12 @@ public class DfsPartitionDataReader extends PartitionDataReader {
   private final FSDataInputStream indexInputStream;
 
   public DfsPartitionDataReader(
-      DiskFileInfo fileInfo, ByteBuffer headerBuffer, ByteBuffer indexBuffer) throws IOException {
+      DiskFileInfo fileInfo,
+      FSDataInputStream dataInputStream,
+      FSDataInputStream indexInputStream,
+      ByteBuffer headerBuffer,
+      ByteBuffer indexBuffer)
+      throws IOException {
     super(fileInfo, headerBuffer, indexBuffer);
     FileSystem fileSystem =
         StorageManager.hadoopFs()
@@ -43,8 +48,8 @@ public class DfsPartitionDataReader extends PartitionDataReader {
                 fileInfo.isHdfs()
                     ? StorageInfo.Type.HDFS
                     : fileInfo.isS3() ? StorageInfo.Type.S3 : StorageInfo.Type.OSS);
-    this.dataInputStream = fileSystem.open(fileInfo.getDfsPath());
-    this.indexInputStream = fileSystem.open(fileInfo.getDfsIndexPath());
+    this.dataInputStream = dataInputStream;
+    this.indexInputStream = indexInputStream;
     this.dataFileSize = fileSystem.getFileStatus(fileInfo.getDfsPath()).getLen();
     this.indexFileSize = fileSystem.getFileStatus(fileInfo.getDfsIndexPath()).getLen();
   }

worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/LocalPartitionDataReader.java

@@ -25,7 +25,6 @@ import io.netty.buffer.ByteBuf;
 import org.apache.commons.io.IOUtils;
 
 import org.apache.celeborn.common.meta.DiskFileInfo;
-import org.apache.celeborn.common.util.FileChannelUtils;
 import org.apache.celeborn.common.util.Utils;
 
 public class LocalPartitionDataReader extends PartitionDataReader {
@@ -34,10 +33,15 @@ public class LocalPartitionDataReader extends PartitionDataReader {
   private final FileChannel indexFileChannel;
 
   public LocalPartitionDataReader(
-      DiskFileInfo fileInfo, ByteBuffer headerBuffer, ByteBuffer indexBuffer) throws IOException {
+      DiskFileInfo fileInfo,
+      FileChannel dataFileChannel,
+      FileChannel indexFileChannel,
+      ByteBuffer headerBuffer,
+      ByteBuffer indexBuffer)
+      throws IOException {
     super(fileInfo, headerBuffer, indexBuffer);
-    this.dataFileChanel = FileChannelUtils.openReadableFileChannel(fileInfo.getFilePath());
-    this.indexFileChannel = FileChannelUtils.openReadableFileChannel(fileInfo.getIndexPath());
+    this.dataFileChanel = dataFileChannel;
+    this.indexFileChannel = indexFileChannel;
     this.dataFileSize = dataFileChanel.size();
     this.indexFileSize = indexFileChannel.size();
   }

worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/MapPartitionData.java

@@ -18,6 +18,7 @@
 package org.apache.celeborn.service.deploy.worker.storage;
 
 import java.io.IOException;
+import java.nio.channels.FileChannel;
 import java.util.List;
 import java.util.PriorityQueue;
 import java.util.concurrent.ConcurrentHashMap;
@@ -27,11 +28,14 @@ import java.util.function.Consumer;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.Channel;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.Path;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import org.apache.celeborn.common.meta.DiskFileInfo;
 import org.apache.celeborn.common.meta.MapFileMeta;
+import org.apache.celeborn.common.util.FileChannelUtils;
 import org.apache.celeborn.common.util.JavaUtils;
 import org.apache.celeborn.common.util.ThreadUtils;
 import org.apache.celeborn.service.deploy.worker.memory.BufferQueue;
@@ -46,6 +50,10 @@ public class MapPartitionData implements MemoryManager.ReadBufferTargetChangeLis
   protected final ExecutorService readExecutor;
   protected final ConcurrentHashMap<Long, MapPartitionDataReader> readers =
       JavaUtils.newConcurrentHashMap();
+  private FileChannel dataFileChanel;
+  private FileChannel indexChannel;
+  private FSDataInputStream hdfsDataInputStream;
+  private FSDataInputStream hdfsIndexInputStream;
   private volatile boolean isReleased = false;
   private final BufferQueue bufferQueue = new BufferQueue();
   private AtomicBoolean bufferQueueInitialized = new AtomicBoolean(false);
@@ -90,6 +98,20 @@ public class MapPartitionData implements MemoryManager.ReadBufferTargetChangeLis
                     String.format("worker-map-partition-%s-reader", mapFileMeta.getMountPoint()),
                     false));
 
+    if (diskFileInfo.isDFS()) {
+      this.hdfsDataInputStream =
+          StorageManager.hadoopFs()
+              .get(diskFileInfo.getStorageType())
+              .open(new Path(diskFileInfo.getFilePath()));
+      this.hdfsIndexInputStream =
+          StorageManager.hadoopFs()
+              .get(diskFileInfo.getStorageType())
+              .open(new Path(diskFileInfo.getIndexPath()));
+
+    } else {
+      this.dataFileChanel = FileChannelUtils.openReadableFileChannel(diskFileInfo.getFilePath());
+      this.indexChannel = FileChannelUtils.openReadableFileChannel(diskFileInfo.getIndexPath());
+    }
     MemoryManager.instance().addReadBufferTargetChangeListener(this);
   }
 
@@ -165,7 +187,7 @@ public class MapPartitionData implements MemoryManager.ReadBufferTargetChangeLis
   }
 
   protected void openReader(MapPartitionDataReader reader) throws IOException {
-    reader.open();
+    reader.open(dataFileChanel, indexChannel, hdfsDataInputStream, hdfsIndexInputStream);
   }
 
   public synchronized void readBuffers() {

worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/MapPartitionDataReader.java

@@ -20,6 +20,7 @@ package org.apache.celeborn.service.deploy.worker.storage;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.ClosedChannelException;
+import java.nio.channels.FileChannel;
 import java.util.ArrayDeque;
 import java.util.Queue;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -30,6 +31,7 @@ import com.google.common.annotations.VisibleForTesting;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFutureListener;
+import org.apache.hadoop.fs.FSDataInputStream;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -126,12 +128,19 @@ public class MapPartitionDataReader implements Comparable<MapPartitionDataReader
     this.readFinished = false;
   }
 
-  public void open() throws IOException {
+  public void open(
+      FileChannel dataFileChannel,
+      FileChannel indexFileChannel,
+      FSDataInputStream dataInputStream,
+      FSDataInputStream indexInputStream)
+      throws IOException {
     if (!isOpen) {
       this.partitionDataReader =
           fileInfo.isDFS()
-              ? new DfsPartitionDataReader(fileInfo, headerBuffer, indexBuffer)
-              : new LocalPartitionDataReader(fileInfo, headerBuffer, indexBuffer);
+              ? new DfsPartitionDataReader(
+                  fileInfo, dataInputStream, indexInputStream, headerBuffer, indexBuffer)
+              : new LocalPartitionDataReader(
+                  fileInfo, dataFileChannel, indexFileChannel, headerBuffer, indexBuffer);
       // index is (offset,length)
       long indexRegionSize = mapFileMeta.getNumSubpartitions() * (long) INDEX_ENTRY_SIZE;
       this.numRegions =