MINOR: Upgrade Hadoop to 3.3.6

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java

@@ -65,6 +65,7 @@
 import org.apache.parquet.HadoopReadOptions;
 import org.apache.parquet.ParquetReadOptions;
 import org.apache.parquet.Preconditions;
+import org.apache.parquet.bytes.ByteBufferAllocator;
 import org.apache.parquet.bytes.ByteBufferInputStream;
 import org.apache.parquet.bytes.ByteBufferReleaser;
 import org.apache.parquet.bytes.BytesInput;
@@ -1378,8 +1379,30 @@ private void readVectored(List<ConsecutivePartList> allParts, ChunkListBuilder b
       totalSize += len;
     }
     LOG.debug("Reading {} bytes of data with vectored IO in {} ranges", totalSize, ranges.size());
-    // Request a vectored read;
-    f.readVectored(ranges, options.getAllocator());
+    // ChecksumFileSystem may allocate internal checksum buffers and return slices of the data buffers.
+    // Capture every original allocation so the row group can release the actual allocator-owned buffers.
+    List<ByteBuffer> allocatedBuffers = new ArrayList<>();
+    ByteBufferAllocator capturingAllocator = new ByteBufferAllocator() {
+      @Override
+      public ByteBuffer allocate(int size) {
+        ByteBuffer buffer = options.getAllocator().allocate(size);
+        allocatedBuffers.add(buffer);
+        return buffer;
+      }
+
+      @Override
+      public void release(ByteBuffer buffer) {
+        options.getAllocator().release(buffer);
+      }
+
+      @Override
+      public boolean isDirect() {
+        return options.getAllocator().isDirect();
+      }
+    };
+    // Request a vectored read.
+    f.readVectored(ranges, capturingAllocator);
+    builder.addBuffersToRelease(allocatedBuffers);
     int k = 0;
     for (ConsecutivePartList consecutivePart : allParts) {
       ParquetFileRange currRange = ranges.get(k++);

parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetFileReaderBufferLeak.java

@@ -169,6 +169,29 @@ public void testCloseReleasesCurrentRowGroupBuffers() throws Exception {
     }
   }
 
+  /**
+   * Verify that closing a reader after a vectored read releases all buffers allocated by the file system.
+   */
+  @Test
+  public void testClosingVectoredReaderReleasesAllBuffers() throws Exception {
+    Path path = writeMultiRowGroupFile(500);
+
+    try (TrackingByteBufferAllocator readAllocator =
+        TrackingByteBufferAllocator.wrap(new HeapByteBufferAllocator())) {
+      ParquetReadOptions options = ParquetReadOptions.builder()
+          .withAllocator(readAllocator)
+          .withUseHadoopVectoredIo(true)
+          .build();
+      InputFile inputFile = HadoopInputFile.fromPath(path, CONF);
+
+      try (ParquetFileReader reader = new ParquetFileReader(inputFile, options)) {
+        PageReadStore pages = reader.readNextRowGroup();
+        assertNotNull(pages);
+        assertTrue(pages.getRowCount() > 0);
+      }
+    }
+  }
+
   /**
    * Verify that readNextFilteredRowGroup() releases buffers of the previous row group
    * when the filter does not trigger column-index filtering (falls back to readNextRowGroup).

pom.xml

@@ -83,7 +83,7 @@
     <spotless.version>3.8.0</spotless.version>
     <shade.prefix>shaded.parquet</shade.prefix>
     <!-- Guarantees no newer classes/methods/constants are used by parquet. -->
-    <hadoop.version>3.3.0</hadoop.version>
+    <hadoop.version>3.3.6</hadoop.version>
     <previous.version>1.18.0</previous.version>
     <thrift.executable>thrift</thrift.executable>
     <format.thrift.executable>${thrift.executable}</format.thrift.executable>