HADOOP-18296. Memory fragmentation in ChecksumFileSystem readVectored() (#7732)

Option "fs.file.checksum.verify" disables checksum
verification in local FS, so sliced subsets of larger buffers are
never returned. 

Stream capability  "fs.capability.vectoredio.sliced" is true
if a filesystem knows that it is returning slices of a larger buffer.
This is false if a filesystem doesn't, or against the local
FS in releases which lack this feature.

Contributed by Steve Loughran

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/ChecksumFileSystem.java

@@ -29,6 +29,7 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.EnumSet;
 import java.util.List;
+import java.util.Locale;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
@@ -57,6 +58,7 @@ import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_ST
 import static org.apache.hadoop.fs.VectoredReadUtils.validateAndSortRanges;
 import static org.apache.hadoop.fs.impl.PathCapabilitiesSupport.validatePathCapabilityArgs;
 import static org.apache.hadoop.fs.impl.StoreImplementationUtils.isProbeForSyncable;
+import static org.apache.hadoop.io.Sizes.S_0;
 
 /****************************************************************
  * Abstract Checksumed FileSystem.
@@ -101,6 +103,14 @@ public abstract class ChecksumFileSystem extends FilterFileSystem {
     this.verifyChecksum = verifyChecksum;
   }
 
+  /**
+   * Is checksum verification enabled?
+   * @return true if files are to be verified through checksums.
+   */
+  public boolean getVerifyChecksum() {
+    return verifyChecksum;
+  }
+
   @Override
   public void setWriteChecksum(boolean writeChecksum) {
     this.writeChecksum = writeChecksum;
@@ -165,7 +175,8 @@ public abstract class ChecksumFileSystem extends FilterFileSystem {
 
   /*******************************************************
    * For open()'s FSInputStream
-   * It verifies that data matches checksums.
+   * It verifies that data matches checksums iff the data
+   * file has matching checksums.
    *******************************************************/
   private static class ChecksumFSInputChecker extends FSInputChecker implements
       IOStatisticsSource, StreamCapabilities {
@@ -426,8 +437,18 @@ public abstract class ChecksumFileSystem extends FilterFileSystem {
       return data;
     }
 
+    /**
+     * Turn off range merging to make buffer recycling more likely (but not guaranteed).
+     * @return 0, always
+     */
+    @Override
+    public int maxReadSizeForVectorReads() {
+      return S_0;
+    }
+
     /**
      * Vectored read.
+     * <p>
      * If the file has no checksums: delegate to the underlying stream.
      * If the file is checksummed: calculate the checksum ranges as
      * well as the data ranges, read both, and validate the checksums
@@ -448,10 +469,12 @@ public abstract class ChecksumFileSystem extends FilterFileSystem {
         final Consumer<ByteBuffer> release) throws IOException {
 
       // If the stream doesn't have checksums, just delegate.
-      if (sums == null) {
+      if (dataStreamToHandleVectorIO()) {
+        LOG.debug("No checksums for vectored read, delegating to inner stream");
         datas.readVectored(ranges, allocate);
         return;
       }
+      LOG.debug("Checksum vectored read for {} ranges", ranges.size());
       final long length = getFileLength();
       final List<? extends FileRange> sorted = validateAndSortRanges(ranges,
           Optional.of(length));
@@ -489,9 +512,37 @@ public abstract class ChecksumFileSystem extends FilterFileSystem {
       }
     }
 
+    /**
+     * Predicate to determine whether vector reads should be directly
+     * handled by the data stream, rather than processing
+     * the ranges in this class, processing which includes checksum validation.
+     * <p>
+     * Vector reading is delegated whenever there are no checksums for
+     * the data file, or when validating checksums has been delegated.
+     * @return true if vector reads are to be directly handled by
+     * the data stream.
+     */
+    private boolean dataStreamToHandleVectorIO() {
+      return sums == null;
+    }
+
+    /**
+     * For this stream, declare that range merging may take place;
+     * otherwise delegate to the inner stream.
+     * @param capability string to query the stream support for.
+     * @return true for sliced vector IO if checksum validation
+     * is taking place. False if no checksums are present for the validation.
+     * For all other probes: pass to the wrapped stream
+     */
     @Override
     public boolean hasCapability(String capability) {
-      return datas.hasCapability(capability);
+      switch (capability.toLowerCase(Locale.ENGLISH)) {
+        // slicing can take place during coalescing and checksumming
+      case StreamCapabilities.VECTOREDIO_BUFFERS_SLICED:
+        return !dataStreamToHandleVectorIO();
+      default:
+        return datas.hasCapability(capability);
+      }
     }
   }
 
@@ -1142,6 +1193,9 @@ public abstract class ChecksumFileSystem extends FilterFileSystem {
     case CommonPathCapabilities.FS_APPEND:
     case CommonPathCapabilities.FS_CONCAT:
       return false;
+    case StreamCapabilities.VECTOREDIO_BUFFERS_SLICED:
+      return getVerifyChecksum();
+
     default:
       return super.hasPathCapability(p, capability);
     }

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/CommonConfigurationKeys.java

@@ -505,4 +505,10 @@ public class CommonConfigurationKeys extends CommonConfigurationKeysPublic {
   public static final String HADOOP_SECURITY_RESOLVER_IMPL =
       "hadoop.security.resolver.impl";
 
+  /**
+   * Verify checksums on read -default is true.
+   * <p>
+   * {@value}.
+   */
+  public static final String LOCAL_FS_VERIFY_CHECKSUM = "fs.file.checksum.verify";
 }

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/LocalFileSystem.java

@@ -27,6 +27,8 @@ import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.conf.Configuration;
 
+import static org.apache.hadoop.fs.CommonConfigurationKeys.LOCAL_FS_VERIFY_CHECKSUM;
+
 /****************************************************************
  * Implement the FileSystem API for the checksumed local filesystem.
  *
@@ -50,6 +52,10 @@ public class LocalFileSystem extends ChecksumFileSystem {
     if (!scheme.equals(fs.getUri().getScheme())) {
       swapScheme = scheme;
     }
+    final boolean checksum = conf.getBoolean(LOCAL_FS_VERIFY_CHECKSUM, true);
+    setVerifyChecksum(checksum);
+    LOG.debug("Checksum verification enabled={}", checksum);
+
   }
 
   /**

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/RawLocalFileSystem.java

@@ -72,6 +72,7 @@ import org.apache.hadoop.util.Shell;
 import org.apache.hadoop.util.StringUtils;
 
 import static org.apache.hadoop.fs.VectoredReadUtils.LOG_BYTE_BUFFER_RELEASED;
+import static org.apache.hadoop.fs.VectoredReadUtils.hasVectorIOCapability;
 import static org.apache.hadoop.fs.VectoredReadUtils.sortRangeList;
 import static org.apache.hadoop.fs.VectoredReadUtils.validateRangeRequest;
 import static org.apache.hadoop.fs.impl.PathCapabilitiesSupport.validatePathCapabilityArgs;
@@ -293,15 +294,12 @@ public class RawLocalFileSystem extends FileSystem {
 
     @Override
     public boolean hasCapability(String capability) {
-      // a bit inefficient, but intended to make it easier to add
-      // new capabilities.
       switch (capability.toLowerCase(Locale.ENGLISH)) {
       case StreamCapabilities.IOSTATISTICS:
       case StreamCapabilities.IOSTATISTICS_CONTEXT:
-      case StreamCapabilities.VECTOREDIO:
         return true;
       default:
-        return false;
+        return hasVectorIOCapability(capability);
       }
     }
 
@@ -400,7 +398,9 @@ public class RawLocalFileSystem extends FileSystem {
       for(int i = 0; i < ranges.size(); ++i) {
         FileRange range = ranges.get(i);
         buffers[i] = allocateRelease.getBuffer(false, range.getLength());
-        channel.read(buffers[i], range.getOffset(), i, this);
+        final ByteBuffer buffer = buffers[i];
+        LOG.debug("Reading file range {} into buffer {}", range, System.identityHashCode(buffer));
+        channel.read(buffer, range.getOffset(), i, this);
       }
     }
 
@@ -416,6 +416,8 @@ public class RawLocalFileSystem extends FileSystem {
     public void completed(Integer result, Integer rangeIndex) {
       FileRange range = ranges.get(rangeIndex);
       ByteBuffer buffer = buffers[rangeIndex];
+      LOG.debug("Completed read range {} into buffer {} outcome={} remaining={}",
+          range, System.identityHashCode(buffer), result, buffer.remaining());
       if (result == -1) {
         // no data was read back.
         failed(new EOFException("Read past End of File"), rangeIndex);

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/StreamCapabilities.java

@@ -86,6 +86,16 @@ public interface StreamCapabilities {
    */
   String VECTOREDIO = "in:readvectored";
 
+  /**
+   * Probe for vector IO implementation details: {@value}.
+   * When performing vectored IO operations, are the buffers returned by readVectored()
+   * potentially sliced subsets of buffers allocated by the allocate() function
+   * passed in the read requests?
+   * If true, this means that the returned buffers may be sliced subsets of the
+   * allocated buffers.
+   */
+  String VECTOREDIO_BUFFERS_SLICED = "fs.capability.vectoredio.sliced";
+
   /**
    * Stream abort() capability implemented by {@link Abortable#abort()}.
    * This matches the Path Capability

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/VectoredReadUtils.java

@@ -24,6 +24,7 @@ import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Comparator;
 import java.util.List;
+import java.util.Locale;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Consumer;
@@ -476,6 +477,22 @@ public final class VectoredReadUtils {
     return readData;
   }
 
+  /**
+   * Default vector IO probes.
+   * These are capabilities which streams that leave vector IO
+   * to the default methods should return when queried for vector capabilities.
+   * @param capability capability to probe for.
+   * @return true if the given capability holds for vectored IO features.
+   */
+  public static boolean hasVectorIOCapability(String capability) {
+    switch (capability.toLowerCase(Locale.ENGLISH)) {
+    case StreamCapabilities.VECTOREDIO:
+      return true;
+    default:
+      return false;
+    }
+  }
+
   /**
    * private constructor.
    */

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/impl/TrackingByteBufferPool.java

@@ -0,0 +1,290 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hadoop.fs.impl;
+
+import java.nio.ByteBuffer;
+import java.util.IdentityHashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.classification.VisibleForTesting;
+import org.apache.hadoop.io.ByteBufferPool;
+
+import static java.lang.System.identityHashCode;
+import static java.util.Objects.requireNonNull;
+
+/**
+ * A wrapper {@link ByteBufferPool} implementation that tracks whether all allocated buffers
+ * are released.
+ * <p>
+ * It throws the related exception at {@link #close()} if any buffer remains un-released.
+ * It also clears the buffers at release so if they continued being used it'll generate errors.
+ * <p>
+ * To be used for testing..
+ * <p>
+ * The stacktraces of the allocation are not stored by default because
+ * it can significantly decrease the unit test performance.
+ * Configuring this class to log at DEBUG will trigger their collection.
+ * @see ByteBufferAllocationStacktraceException
+ * <p>
+ * Adapted from Parquet class {@code org.apache.parquet.bytes.TrackingByteBufferAllocator}.
+ */
+@VisibleForTesting
+public final class TrackingByteBufferPool implements ByteBufferPool, AutoCloseable {
+
+  private static final Logger LOG = LoggerFactory.getLogger(TrackingByteBufferPool.class);
+
+  /**
+   * Wrap an existing allocator with this tracking allocator.
+   * @param allocator allocator to wrap.
+   * @return a new allocator.
+   */
+  public static TrackingByteBufferPool wrap(ByteBufferPool allocator) {
+    return new TrackingByteBufferPool(allocator);
+  }
+
+  public static class LeakDetectorHeapByteBufferPoolException
+      extends RuntimeException {
+
+    private LeakDetectorHeapByteBufferPoolException(String msg) {
+      super(msg);
+    }
+
+    private LeakDetectorHeapByteBufferPoolException(String msg, Throwable cause) {
+      super(msg, cause);
+    }
+
+    private LeakDetectorHeapByteBufferPoolException(
+        String message,
+        Throwable cause,
+        boolean enableSuppression,
+        boolean writableStackTrace) {
+      super(message, cause, enableSuppression, writableStackTrace);
+    }
+  }
+
+  /**
+   * Strack trace of allocation as saved in the tracking map.
+   */
+  public static final class ByteBufferAllocationStacktraceException
+      extends LeakDetectorHeapByteBufferPoolException {
+
+    /**
+     * Single stack trace instance to use when DEBUG is not enabled.
+     */
+    private static final ByteBufferAllocationStacktraceException WITHOUT_STACKTRACE =
+        new ByteBufferAllocationStacktraceException(false);
+
+    /**
+     * Create a stack trace for the map, either using the shared static one
+     * or a dynamically created one.
+     * @return a stack
+     */
+    private static ByteBufferAllocationStacktraceException create() {
+      return LOG.isDebugEnabled()
+          ? new ByteBufferAllocationStacktraceException()
+          : WITHOUT_STACKTRACE;
+    }
+
+    private ByteBufferAllocationStacktraceException() {
+      super("Allocation stacktrace of the first ByteBuffer:");
+    }
+
+    /**
+     * Private constructor to for the singleton {@link #WITHOUT_STACKTRACE},
+     * telling develoers how to see a trace per buffer.
+     */
+    private ByteBufferAllocationStacktraceException(boolean unused) {
+      super("Log org.apache.hadoop.fs.impl.TrackingByteBufferPool at DEBUG for stack traces",
+          null,
+          false,
+          false);
+    }
+  }
+
+  /**
+   * Exception raised in {@link TrackingByteBufferPool#putBuffer(ByteBuffer)} if the
+   * buffer to release was not in the hash map.
+   */
+  public static final class ReleasingUnallocatedByteBufferException
+      extends LeakDetectorHeapByteBufferPoolException {
+
+    private ReleasingUnallocatedByteBufferException(final ByteBuffer b) {
+      super(String.format("Releasing a ByteBuffer instance that is not allocated"
+          + " by this buffer pool or already been released: %s size %d; hash code %s",
+          b, b.capacity(), identityHashCode(b)));
+    }
+  }
+
+  /**
+   * Exception raised in {@link TrackingByteBufferPool#close()} if there
+   * was an unreleased buffer.
+   */
+  public static final class LeakedByteBufferException
+      extends LeakDetectorHeapByteBufferPoolException {
+
+    private final int count;
+
+    private LeakedByteBufferException(int count, ByteBufferAllocationStacktraceException e) {
+      super(count + " ByteBuffer object(s) is/are remained unreleased"
+          + " after closing this buffer pool.", e);
+      this.count = count;
+    }
+
+    /**
+     * Get the number of unreleased buffers.
+     * @return number of unreleased buffers
+     */
+    public int getCount() {
+      return count;
+    }
+  }
+
+  /**
+   * Tracker of allocations.
+   * <p>
+   * The key maps by the object id of the buffer, and refers to either a common stack trace
+   * or one dynamically created for each allocation.
+   */
+  private final Map<ByteBuffer, ByteBufferAllocationStacktraceException> allocated =
+      new IdentityHashMap<>();
+
+  /**
+   * Wrapped buffer pool.
+   */
+  private final ByteBufferPool allocator;
+
+  /**
+   * Number of buffer allocations.
+   * <p>
+   * This is incremented in {@link #getBuffer(boolean, int)}.
+   */
+  private final AtomicInteger bufferAllocations = new AtomicInteger();
+
+  /**
+   * Number of buffer releases.
+   * <p>
+   * This is incremented in {@link #putBuffer(ByteBuffer)}.
+   */
+  private final AtomicInteger bufferReleases = new AtomicInteger();
+
+  /**
+   * private constructor.
+   * @param allocator pool allocator.
+   */
+  private TrackingByteBufferPool(ByteBufferPool allocator) {
+    this.allocator = allocator;
+  }
+
+  public int getBufferAllocations() {
+    return bufferAllocations.get();
+  }
+
+  public int getBufferReleases() {
+    return bufferReleases.get();
+  }
+
+  /**
+   * Get a buffer from the pool.
+   * <p>
+   * This increments the {@link #bufferAllocations} counter and stores the
+   * singleron or local allocation stack trace in the {@link #allocated} map.
+   * @param direct whether to allocate a direct buffer or not
+   * @param size size of the buffer to allocate
+   * @return a ByteBuffer instance
+   */
+  @Override
+  public synchronized ByteBuffer getBuffer(final boolean direct, final int size) {
+    bufferAllocations.incrementAndGet();
+    ByteBuffer buffer = allocator.getBuffer(direct, size);
+    final ByteBufferAllocationStacktraceException ex =
+        ByteBufferAllocationStacktraceException.create();
+    allocated.put(buffer, ex);
+    LOG.debug("Creating ByteBuffer:{} size {} {}",
+        identityHashCode(buffer), size, buffer, ex);
+    return buffer;
+  }
+
+  /**
+   * Release a buffer back to the pool.
+   * <p>
+   * This increments the {@link #bufferReleases} counter and removes the
+   * buffer from the {@link #allocated} map.
+   * <p>
+   * If the buffer was not allocated by this pool, it throws
+   * {@link ReleasingUnallocatedByteBufferException}.
+   *
+   * @param buffer buffer to release
+   * @throws ReleasingUnallocatedByteBufferException if the buffer was not allocated by this pool
+   */
+  @Override
+  public synchronized void putBuffer(ByteBuffer buffer)
+      throws ReleasingUnallocatedByteBufferException {
+
+    bufferReleases.incrementAndGet();
+    requireNonNull(buffer);
+    LOG.debug("Releasing ByteBuffer: {}: {}", identityHashCode(buffer), buffer);
+    if (allocated.remove(buffer) == null) {
+      throw new ReleasingUnallocatedByteBufferException(buffer);
+    }
+    allocator.putBuffer(buffer);
+    // Clearing the buffer so subsequent access would probably generate errors
+    buffer.clear();
+  }
+
+  /**
+   * Check if the buffer is in the pool.
+   * @param buffer buffer
+   * @return true if the buffer is in the pool
+   */
+  public boolean containsBuffer(ByteBuffer buffer) {
+    return allocated.containsKey(requireNonNull(buffer));
+  }
+
+  /**
+   * Get the number of allocated buffers.
+   * @return number of allocated buffers
+   */
+  public int size() {
+    return allocated.size();
+  }
+
+  /**
+   * Expect all buffers to be released -if not, log unreleased ones
+   * and then raise an exception with the stack trace of the first
+   * unreleased buffer.
+   * @throws LeakedByteBufferException if at least one buffer was not released
+   */
+  @Override
+  public void close() throws LeakedByteBufferException {
+    if (!allocated.isEmpty()) {
+      allocated.keySet().forEach(buffer ->
+          LOG.warn("Unreleased ByteBuffer {}; {}", identityHashCode(buffer), buffer));
+      LeakedByteBufferException ex = new LeakedByteBufferException(
+          allocated.size(),
+          allocated.values().iterator().next());
+      allocated.clear(); // Drop the references to the ByteBuffers, so they can be gc'd
+      throw ex;
+    }
+  }
+}

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/Sizes.java

@@ -31,6 +31,9 @@ import org.apache.hadoop.classification.InterfaceStability;
 @InterfaceStability.Evolving
 public final class Sizes {
 
+  /** 0 bytes: {@value}. Here to make it easy to find use of zero in constants. */
+  public static final int S_0 = 0;
+
   /** 2^8 bytes: {@value}. */
   public static final int S_256 = 256;
 

hadoop-common-project/hadoop-common/src/main/resources/core-default.xml

@@ -1364,6 +1364,24 @@
   <description>File space usage statistics refresh interval in msec.</description>
 </property>
 
+<property>
+  <name>fs.file.checksum.verify</name>
+  <value>true</value>
+  <description>
+    Should data read through the local filesystem (file://) URLs be verified aginst
+    the checksums stored in the associated checksum files?
+    Setting this to false skips loading the checksum files, reading data in checksum-aligned
+    blocks and verifying checksums. This may improve performance
+    when reading data, though it pushes the responsibility of detecting errors
+    into the file formats themselves, or the underlying storage system.
+    Even when verification is enabled, files without associated checksum files
+    .$FILENAME.crc are never verified.
+    When fs.file.checksum.verify is false, vector reads of data will always return
+    buffers that are the buffers allocated through the buffer allocator
+    passed in to the API call and not sliced subsets thereof.
+  </description>
+</property>
+
 <property>
   <name>fs.automatic.close</name>
   <value>true</value>

hadoop-common-project/hadoop-common/src/site/markdown/filesystem/fsdatainputstream.md

@@ -541,7 +541,7 @@ end of first and start of next range is more than this value.
 #### `maxReadSizeForVectorReads()`
 
 Maximum number of bytes which can be read in one go after merging the ranges.
-Two ranges won't be merged if the combined data to be read It's okay we have a look at what we do right now for readOkayis more than this value.
+Two ranges won't be merged if the combined data to be read.
 Essentially setting this to 0 will disable the merging of ranges.
 
 #### Concurrency
@@ -647,7 +647,7 @@ For details see [HADOOP-19291](https://issues.apache.org/jira/browse/HADOOP-1929
 For reliable use with older hadoop releases with the API: sort the list of ranges
 and check for overlaps before calling `readVectored()`.
 
-*Direct Buffer Reads*
+#### Direct Buffer Reads
 
 Releases without [HADOOP-19101](https://issues.apache.org/jira/browse/HADOOP-19101)
 _Vectored Read into off-heap buffer broken in fallback implementation_ can read data
@@ -665,8 +665,54 @@ support through an explicit `hasCapability()` probe:
 Stream.hasCapability("in:readvectored")
 ```
 
-Given the HADOOP-18296 problem with `ChecksumFileSystem` and direct buffers, across all releases,
-it is best to avoid using this API in production with direct buffers.
+#### Buffer Slicing
+
+[HADOOP-18296](https://issues.apache.org/jira/browse/HADOOP-18296),
+_Memory fragmentation in ChecksumFileSystem Vectored IO implementation_
+highlights that `ChecksumFileSystem` (which the default implementation of `file://`
+subclasses), may return buffers which are sliced subsets of buffers allocated
+through the `allocate()` function passed in.
+
+This will happen during reads with and without range coalescing.
+
+Checksum verification may be disabled by setting the option
+`fs.file.checksum.verify` to false (Hadoop 3.4.2 and later).
+
+```xml
+<property>
+  <name>fs.file.checksum.verify</name>
+  <value>false</value>
+</property>
+```
+
+(As you would expect, disabling checksum verification means that errors
+reading data may not be detected during the read operation.
+Use with care in production.)
+
+Filesystem instances which split buffers during vector read operations
+MUST declare this by returning `true`
+to the path capabilities probe `fs.capability.vectoredio.sliced`,
+and for the open stream in its `hasCapability()` method.
+
+
+The local filesystem will not slice buffers if the checksum file
+of `filename + ".crc"` is not found. This is not declared in the
+filesystem `hasPathCapability(filename, "fs.capability.vectoredio.sliced")`
+call, as no checks for the checksum file are made then.
+This cannot be relied on in production, but it may be useful when
+testing for buffer recycling with Hadoop releases 3.4.1 and earlier.
+
+*Implementors Notes*
+
+* Don't slice buffers. `ChecksumFileSystem` has to be considered an outlier which
+  needs to be addressed in future.
+* Always free buffers in error handling code paths.
+* When handling errors in coalesced ranges, don't release buffers for any sub-ranges
+  which have already completed.
+
+Handling failures in coalesced ranges is complicated. Recent implementations, such as
+`org.apache.hadoop.fs.s3a.impl.streams.AnalyticsStream` omit range coalescing,
+relying solely on parallel HTTP for performance.
 
 
 ## `void readVectored(List<? extends FileRange> ranges, IntFunction<ByteBuffer> allocate, Consumer<ByteBuffer> release)`

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/AbstractContractVectoredReadTest.java

@@ -44,6 +44,7 @@ import org.apache.hadoop.fs.FileRange;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.impl.TrackingByteBufferPool;
 import org.apache.hadoop.io.ElasticByteBufferPool;
 import org.apache.hadoop.io.WeakReferencedElasticByteBufferPool;
 import org.apache.hadoop.util.concurrent.HadoopExecutors;
@@ -52,13 +53,16 @@ import static java.util.Arrays.asList;
 import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_LENGTH;
 import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY;
 import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY_VECTOR;
+import static org.apache.hadoop.fs.StreamCapabilities.VECTOREDIO_BUFFERS_SLICED;
 import static org.apache.hadoop.fs.contract.ContractTestUtils.VECTORED_READ_OPERATION_TEST_TIMEOUT_SECONDS;
 import static org.apache.hadoop.fs.contract.ContractTestUtils.assertDatasetEquals;
 import static org.apache.hadoop.fs.contract.ContractTestUtils.createFile;
 import static org.apache.hadoop.fs.contract.ContractTestUtils.range;
 import static org.apache.hadoop.fs.contract.ContractTestUtils.returnBuffersToPoolPostRead;
 import static org.apache.hadoop.fs.contract.ContractTestUtils.validateVectoredReadResult;
-  import static org.apache.hadoop.test.LambdaTestUtils.intercept;
+import static org.apache.hadoop.io.Sizes.S_128K;
+import static org.apache.hadoop.io.Sizes.S_4K;
+import static org.apache.hadoop.test.LambdaTestUtils.intercept;
 import static org.apache.hadoop.test.LambdaTestUtils.interceptFuture;
 import static org.apache.hadoop.util.functional.FutureIO.awaitFuture;
 
@@ -74,7 +78,7 @@ public abstract class AbstractContractVectoredReadTest extends AbstractFSContrac
   private static final Logger LOG =
       LoggerFactory.getLogger(AbstractContractVectoredReadTest.class);
 
-  public static final int DATASET_LEN = 64 * 1024;
+  public static final int DATASET_LEN = S_128K;
   protected static final byte[] DATASET = ContractTestUtils.dataset(DATASET_LEN, 'a', 32);
   protected static final String VECTORED_READ_FILE_NAME = "vectored_file.txt";
 
@@ -91,6 +95,8 @@ public abstract class AbstractContractVectoredReadTest extends AbstractFSContrac
 
   private final String bufferType;
 
+  private final boolean isDirect;
+
   /**
    * Path to the vector file.
    */
@@ -110,7 +116,7 @@ public abstract class AbstractContractVectoredReadTest extends AbstractFSContrac
 
   protected AbstractContractVectoredReadTest(String bufferType) {
     this.bufferType = bufferType;
-    final boolean isDirect = !"array".equals(bufferType);
+    this.isDirect = !"array".equals(bufferType);
     this.allocate = size -> pool.getBuffer(isDirect, size);
   }
 
@@ -619,4 +625,61 @@ public abstract class AbstractContractVectoredReadTest extends AbstractFSContrac
       });
     }
   }
+
+  @Test
+  public void testBufferSlicing() throws Throwable {
+    describe("Test buffer slicing behavior in vectored IO");
+
+    final int numBuffers = 8;
+    final int bufferSize = S_4K;
+    long offset = 0;
+    final List<FileRange> fileRanges = new ArrayList<>();
+    for (int i = 0; i < numBuffers; i++) {
+      fileRanges.add(FileRange.createFileRange(offset, bufferSize));
+      // increment and add a non-binary-aligned gap, so as to force
+      // offsets to be misaligned with possible page sizes.
+      offset += bufferSize + 4000;
+    }
+    TrackingByteBufferPool trackerPool = TrackingByteBufferPool.wrap(getPool());
+    int unknownBuffers = 0;
+    boolean slicing;
+    try (FSDataInputStream in = openVectorFile()) {
+      slicing = in.hasCapability(VECTOREDIO_BUFFERS_SLICED);
+      LOG.info("Slicing is {} for vectored IO with stream {}", slicing, in);
+      in.readVectored(fileRanges, s -> trackerPool.getBuffer(isDirect, s), trackerPool::putBuffer);
+
+      // check that all buffers are from the the pool, unless they are sliced.
+      for (FileRange res : fileRanges) {
+        CompletableFuture<ByteBuffer> data = res.getData();
+        ByteBuffer buffer = awaitFuture(data);
+        Assertions.assertThat(buffer)
+            .describedAs("Buffer must not be null")
+            .isNotNull();
+        Assertions.assertThat(slicing || trackerPool.containsBuffer(buffer))
+            .describedAs("Buffer must be from the pool")
+            .isTrue();
+        try {
+          trackerPool.putBuffer(buffer);
+        } catch (TrackingByteBufferPool.ReleasingUnallocatedByteBufferException e) {
+          // this can happen if the buffer was sliced, as it is not in the pool.
+          if (!slicing) {
+            throw e;
+          }
+          LOG.info("Sliced buffer detected: {}", buffer);
+          unknownBuffers++;
+        }
+      }
+    }
+    try {
+      trackerPool.close();
+    } catch (TrackingByteBufferPool.LeakedByteBufferException e) {
+      if (!slicing) {
+        throw e;
+      }
+      LOG.info("Slicing is enabled; we saw leaked buffers: {} after {}"
+              + " releases of unknown buffers",
+          e.getCount(), unknownBuffers);
+    }
+
+  }
 }

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/localfs/TestLocalFSContractVectoredRead.java

@@ -74,6 +74,13 @@ public class TestLocalFSContractVectoredRead extends AbstractContractVectoredRea
     validateCheckReadException(testPath, length, smallFileRanges);
   }
 
+  /**
+   * Verify that checksum validation works through vectored reads.
+   * @param testPath path to the file to be tested
+   * @param length length of the file to be created
+   * @param ranges ranges to be read from the file
+   * @throws Exception any exception other than ChecksumException
+   */
   private void validateCheckReadException(Path testPath,
                                           int length,
                                           List<FileRange> ranges) throws Exception {

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/viewfs/TestViewFileSystemDelegation.java

@@ -157,7 +157,8 @@ public class TestViewFileSystemDelegation { //extends ViewFileSystemTestSetup {
     public void setVerifyChecksum(boolean verifyChecksum) {
       this.verifyChecksum = verifyChecksum;
     }
-    
+
+    @Override
     public boolean getVerifyChecksum(){
       return verifyChecksum;
     }

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/contract/s3a/ITestS3AContractVectoredRead.java

@@ -62,7 +62,7 @@ import static org.apache.hadoop.fs.contract.ContractTestUtils.returnBuffersToPoo
 import static org.apache.hadoop.fs.contract.ContractTestUtils.validateVectoredReadResult;
 import static org.apache.hadoop.fs.s3a.Constants.AWS_S3_VECTOR_READS_MAX_MERGED_READ_SIZE;
 import static org.apache.hadoop.fs.s3a.Constants.AWS_S3_VECTOR_READS_MIN_SEEK_SIZE;
-import static org.apache.hadoop.fs.s3a.S3ATestUtils.skipIfAnalyticsAcceleratorEnabled;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.disableAnalyticsAccelerator;
 import static org.apache.hadoop.fs.statistics.IOStatisticAssertions.verifyStatisticCounterValue;
 import static org.apache.hadoop.fs.statistics.IOStatisticsLogging.ioStatisticsToPrettyString;
 import static org.apache.hadoop.io.Sizes.S_1M;
@@ -71,7 +71,8 @@ import static org.apache.hadoop.test.LambdaTestUtils.interceptFuture;
 import static org.apache.hadoop.test.MoreAsserts.assertEqual;
 
 /**
- * S3A contract tests for vectored reads.
+ * S3A contract tests for vectored reads through the classic input stream.
+ * <p>
  * This is a complex suite as it really is testing the store, so measurements of
  * what IO took place is also performed if the input stream is suitable for this.
  */
@@ -89,14 +90,12 @@ public class ITestS3AContractVectoredRead extends AbstractContractVectoredReadTe
   }
 
   /**
-   * Analytics Accelerator Library for Amazon S3 does not support Vectored Reads.
-   * @throws Exception
+   * Create a configuration.
+   * @return a configuration
    */
   @Override
-  public void setup() throws Exception {
-    super.setup();
-    skipIfAnalyticsAcceleratorEnabled(getContract().getConf(),
-        "Analytics Accelerator does not support vectored reads");
+  protected Configuration createConfiguration() {
+    return disableAnalyticsAccelerator(super.createConfiguration());
   }
 
   /**

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/S3ATestUtils.java

@@ -108,6 +108,7 @@ import static org.apache.hadoop.fs.contract.ContractTestUtils.createFile;
 import static org.apache.hadoop.fs.impl.FlagSet.createFlagSet;
 import static org.apache.hadoop.fs.s3a.S3AEncryptionMethods.SSE_S3;
 import static org.apache.hadoop.fs.s3a.impl.streams.InputStreamType.Analytics;
+import static org.apache.hadoop.fs.s3a.impl.streams.InputStreamType.Classic;
 import static org.apache.hadoop.fs.s3a.impl.streams.InputStreamType.Prefetch;
 import static org.apache.hadoop.fs.s3a.impl.CallableSupplier.submit;
 import static org.apache.hadoop.fs.s3a.impl.CallableSupplier.waitForCompletion;
@@ -1876,6 +1877,18 @@ public final class S3ATestUtils {
     return conf;
   }
 
+  /**
+   * Disable analytics stream for S3A S3AFileSystem in tests.
+   * @param conf Configuration to update
+   * @return patched config
+   */
+  public static Configuration disableAnalyticsAccelerator(Configuration conf) {
+    removeBaseAndBucketOverrides(conf,
+        INPUT_STREAM_TYPE);
+    conf.setEnum(INPUT_STREAM_TYPE, Classic);
+    return conf;
+  }
+
   /**
    * Probe for a filesystem having a specific stream type;
    * this is done through filesystem capabilities.