From 55cf7e56577120d8a4306b8dfd5881b70e6deabe Mon Sep 17 00:00:00 2001
From: Rui Li <lirui@apache.org>
Date: Tue, 28 Apr 2026 18:55:48 +0800
Subject: [PATCH] HDFS-17914. Prevent DFSInputStream from issuing 0-byte reads

---
 .../apache/hadoop/hdfs/DFSInputStream.java    |  4 +-
 .../hadoop/hdfs/TestDFSInputStream.java       | 44 +++++++++++++++++++
 2 files changed, 47 insertions(+), 1 deletion(-)

diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSInputStream.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSInputStream.java
index fa5c3127d3992e..8513a1c3d48055 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSInputStream.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSInputStream.java
@@ -1230,7 +1230,9 @@ void actualGetFromOneDataNode(final DNAddrPair datanode, final long startInBlk,
         long beginReadMS = Time.monotonicNow();
         int nread = 0;
         int ret;
-        while (true) {
+        // Stop once the slice is filled; an extra read with remaining()==0
+        // can trigger wasted slow-lane I/O in SCR.
+        while (tmp.hasRemaining()) {
           ret = reader.read(tmp);
           if (ret <= 0) {
             break;
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSInputStream.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSInputStream.java
index dad93c85dd30a4..0c9804d9940ab1 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSInputStream.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSInputStream.java
@@ -23,11 +23,16 @@
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assumptions.assumeTrue;
+import static org.mockito.AdditionalAnswers.delegatesTo;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 import java.io.File;
 import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Random;
@@ -364,4 +369,43 @@ public Void answer(InvocationOnMock invocation) throws Throwable {
       IOUtils.closeStream(out);
     }
   }
+
+  @Test
+  @Timeout(value = 60)
+  public void testAvoidsEmptyBufferRead() throws IOException {
+    Configuration conf = new Configuration();
+    try (MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf).build()) {
+      DistributedFileSystem fs = cluster.getFileSystem();
+      DFSClient client = fs.getClient();
+      Path file = new Path("/preadFile");
+      int fileLength = 8192;
+      DFSTestUtil.createFile(fs, file, fileLength, (short) 1, 0L);
+
+      AtomicInteger emptyBufferReads = new AtomicInteger(0);
+      try (DFSInputStream in = new DFSInputStream(client, file.toString(),
+          true, null) {
+        @Override
+        protected BlockReader getBlockReader(LocatedBlock targetBlock,
+            long offsetInBlock, long length, InetSocketAddress targetAddr,
+            StorageType storageType, DatanodeInfo datanode)
+            throws IOException {
+          BlockReader real = super.getBlockReader(targetBlock, offsetInBlock,
+              length, targetAddr, storageType, datanode);
+          BlockReader wrapped = mock(BlockReader.class, delegatesTo(real));
+          doAnswer(inv -> {
+            if (!((ByteBuffer) inv.getArgument(0)).hasRemaining()) {
+              emptyBufferReads.incrementAndGet();
+            }
+            return real.read(inv.getArgument(0));
+          }).when(wrapped).read(any(ByteBuffer.class));
+          return wrapped;
+        }
+      }) {
+        byte[] dest = new byte[fileLength];
+        assertEquals(fileLength, in.read(0L, dest, 0, dest.length));
+        assertEquals(0, emptyBufferReads.get(),
+            "DFSInputStream should not call BlockReader.read with an empty buffer");
+      }
+    }
+  }
 }