HDFS-17965. DFSInputStream.close() should release the block reader even if the DFSClient is already closed

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSInputStream.java

@@ -733,26 +733,33 @@ public synchronized void close() throws IOException {
         DFSClient.LOG.debug("DFSInputStream has been closed already");
         return;
       }
-      dfsClient.checkOpen();
-
-      if ((extendedReadBuffers != null) && (!extendedReadBuffers.isEmpty())) {
-        final StringBuilder builder = new StringBuilder();
-        extendedReadBuffers
-            .visitAll(new IdentityHashStore.Visitor<ByteBuffer, Object>() {
-              private String prefix = "";
-
-              @Override
-              public void accept(ByteBuffer k, Object v) {
-                builder.append(prefix).append(k);
-                prefix = ", ";
-              }
-            });
-        DFSClient.LOG.warn("closing file " + src + ", but there are still "
-            + "unreleased ByteBuffers allocated by read().  "
-            + "Please release " + builder.toString() + ".");
+      try {
+        dfsClient.checkOpen();
+
+        if ((extendedReadBuffers != null) && (!extendedReadBuffers.isEmpty())) {
+          final StringBuilder builder = new StringBuilder();
+          extendedReadBuffers
+              .visitAll(new IdentityHashStore.Visitor<ByteBuffer, Object>() {
+                private String prefix = "";
+
+                @Override
+                public void accept(ByteBuffer k, Object v) {
+                  builder.append(prefix).append(k);
+                  prefix = ", ";
+                }
+              });
+          DFSClient.LOG.warn("closing file " + src + ", but there are still "
+              + "unreleased ByteBuffers allocated by read().  "
+              + "Please release " + builder.toString() + ".");
+        }
+      } finally {
+        // Release the block reader even if checkOpen() throws because the
+        // DFSClient was closed first. Otherwise the block reader's socket
+        // leaks and the DataNode side of the connection can be stuck in
+        // FIN_WAIT1 for the lifetime of the client JVM.
+        closeCurrentBlockReaders();
+        super.close();
       }
-      closeCurrentBlockReaders();
-      super.close();
     } finally {
       /**
        * If dfsInputStream is closed and datanode is in
@@ -1826,6 +1833,11 @@ protected void closeCurrentBlockReaders() {
     blockEnd = -1;
   }
 
+  @VisibleForTesting
+  BlockReader getCurrentBlockReader() {
+    return blockReader;
+  }
+
   @Override
   public synchronized void setReadahead(Long readahead)
       throws IOException {

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

@@ -21,6 +21,7 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assumptions.assumeTrue;
 import static org.mockito.Mockito.mock;
@@ -364,4 +365,34 @@ public Void answer(InvocationOnMock invocation) throws Throwable {
       IOUtils.closeStream(out);
     }
   }
+
+  @Test
+  @Timeout(30)
+  public void testCloseReleasesBlockReaderWhenClientAlreadyClosed()
+      throws Exception {
+    Configuration conf = new HdfsConfiguration();
+    try (MiniDFSCluster cluster =
+        new MiniDFSCluster.Builder(conf).numDataNodes(1).build()) {
+      cluster.waitActive();
+      DistributedFileSystem fs = cluster.getFileSystem();
+      Path file = new Path("/testfile");
+      DFSTestUtil.createFile(fs, file, 4096, (short) 1, 0);
+
+      DFSClient client = new DFSClient(cluster.getURI(), conf);
+      DFSInputStream in = client.open("/testfile");
+      assertTrue(in.read() != -1);
+      assertNotNull(in.getCurrentBlockReader());
+
+      // Close the client before the stream. close() must still release the
+      // block reader (and its socket to the DataNode) even though
+      // checkOpen() fails with "Filesystem closed".
+      client.close();
+      try {
+        in.close();
+      } catch (IOException e) {
+        GenericTestUtils.assertExceptionContains("Filesystem closed", e);
+      }
+      assertNull(in.getCurrentBlockReader());
+    }
+  }
 }