WIP HBASE-28065 Corrupt HFile data is mishandled in several cases

* when no block size is provided and there's not a preread headerBuf, treat the value with
  caution.
* verify HBase checksums before making use of the block header.
* inline verifyOnDiskSizeMatchesHeader to keep throw/return logic in the method body.
* whenever a read is determined to be corrupt and fallback to HDFS checksum is necessary, also
  invalidate the cached value of headerBuf.

Author: ndimiduk

hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileBlock.java

@@ -390,12 +390,12 @@ static HFileBlock createFromBuff(ByteBuff buf, boolean usesHBaseChecksum, final
 
   /**
    * Parse total on disk size including header and checksum.
-   * @param headerBuf      Header ByteBuffer. Presumed exact size of header.
-   * @param verifyChecksum true if checksum verification is in use.
+   * @param headerBuf       Header ByteBuffer. Presumed exact size of header.
+   * @param checksumSupport true if checksum verification is in use.
    * @return Size of the block with header included.
    */
-  private static int getOnDiskSizeWithHeader(final ByteBuff headerBuf, boolean verifyChecksum) {
-    return headerBuf.getInt(Header.ON_DISK_SIZE_WITHOUT_HEADER_INDEX) + headerSize(verifyChecksum);
+  private static int getOnDiskSizeWithHeader(final ByteBuff headerBuf, boolean checksumSupport) {
+    return headerBuf.getInt(Header.ON_DISK_SIZE_WITHOUT_HEADER_INDEX) + headerSize(checksumSupport);
   }
 
   /**
@@ -1568,33 +1568,47 @@ public HFileBlock readBlockData(long offset, long onDiskSizeWithHeaderL, boolean
     }
 
     /**
-     * Returns Check <code>onDiskSizeWithHeaderL</code> size is healthy and then return it as an int
+     * Check that {@code value} value seems reasonable, within a large margin of error.
+     * @return {@code true} if the value is safe to proceed, {@code false} otherwise.
      */
-    private static int checkAndGetSizeAsInt(final long onDiskSizeWithHeaderL, final int hdrSize)
-      throws IOException {
-      if (
-        (onDiskSizeWithHeaderL < hdrSize && onDiskSizeWithHeaderL != -1)
-          || onDiskSizeWithHeaderL >= Integer.MAX_VALUE
-      ) {
-        throw new IOException(
-          "Invalid onDisksize=" + onDiskSizeWithHeaderL + ": expected to be at least " + hdrSize
-            + " and at most " + Integer.MAX_VALUE + ", or -1");
+    private boolean checkOnDiskSizeWithHeader(int value) {
+      if (value == -1) {
+        // a magic value we expect to see.
+        return true;
+      }
+      if (value < 0) {
+        if (LOG.isTraceEnabled()) {
+          LOG.trace(
+            "onDiskSizeWithHeader={}; value represents a size, so it should never be negative.",
+            value);
+        }
+        return false;
+      }
+      if (value - hdrSize < 0) {
+        if (LOG.isTraceEnabled()) {
+          LOG.trace(
+            "onDiskSizeWithHeader={}, hdrSize={}; don't accept a value that is negative after the header size is excluded.",
+            value, hdrSize);
+        }
+        return false;
       }
-      return (int) onDiskSizeWithHeaderL;
+      return true;
     }
 
     /**
-     * Verify the passed in onDiskSizeWithHeader aligns with what is in the header else something is
-     * not right.
+     * Check that {@code value} value seems reasonable, within a large margin of error.
+     * @return {@code true} if the value is safe to proceed, {@code false} otherwise.
      */
-    private void verifyOnDiskSizeMatchesHeader(final int passedIn, final ByteBuff headerBuf,
-      final long offset, boolean verifyChecksum) throws IOException {
-      // Assert size provided aligns with what is in the header
-      int fromHeader = getOnDiskSizeWithHeader(headerBuf, verifyChecksum);
-      if (passedIn != fromHeader) {
-        throw new IOException("Passed in onDiskSizeWithHeader=" + passedIn + " != " + fromHeader
-          + ", offset=" + offset + ", fileContext=" + this.fileContext);
+    private boolean checkOnDiskSizeWithHeader(long value) {
+      // same validation logic as is used by Math.toIntExact(long)
+      int intValue = (int) value;
+      if (intValue != value) {
+        if (LOG.isTraceEnabled()) {
+          LOG.trace("onDiskSizeWithHeader={}; value exceeds int size limits.", value);
+        }
+        return false;
       }
+      return checkOnDiskSizeWithHeader(intValue);
     }
 
     /**
@@ -1625,6 +1639,13 @@ private void cacheNextBlockHeader(final long offset, ByteBuff onDiskBlock,
       this.prefetchedHeader.set(ph);
     }
 
+    /**
+     * Clear the cached value when its integrity is suspect.
+     */
+    private void invalidateNextBlockHeader() {
+      prefetchedHeader.set(new PrefetchedHeader());
+    }
+
     private int getNextBlockOnDiskSize(boolean readNextHeader, ByteBuff onDiskBlock,
       int onDiskSizeWithHeader) {
       int nextBlockOnDiskSize = -1;
@@ -1667,7 +1688,11 @@ protected HFileBlock readBlockDataInternal(FSDataInputStream is, long offset,
       final AttributesBuilder attributesBuilder = Attributes.builder();
       Optional.of(Context.current()).map(val -> val.get(CONTEXT_KEY))
         .ifPresent(c -> c.accept(attributesBuilder));
-      int onDiskSizeWithHeader = checkAndGetSizeAsInt(onDiskSizeWithHeaderL, hdrSize);
+      if (!checkOnDiskSizeWithHeader(onDiskSizeWithHeaderL)) {
+        throw new IOException(
+          "Caller provided invalid onDiskSizeWithHeaderL=" + onDiskSizeWithHeaderL);
+      }
+      int onDiskSizeWithHeader = (int) onDiskSizeWithHeaderL;
       // Try and get cached header. Will serve us in rare case where onDiskSizeWithHeaderL is -1
       // and will save us having to seek the stream backwards to reread the header we
       // read the last time through here.
@@ -1682,8 +1707,8 @@ protected HFileBlock readBlockDataInternal(FSDataInputStream is, long offset,
       // file has support for checksums (version 2+).
       boolean checksumSupport = this.fileContext.isUseHBaseChecksum();
       long startTime = EnvironmentEdgeManager.currentTime();
-      if (onDiskSizeWithHeader <= 0) {
-        // We were not passed the block size. Need to get it from the header. If header was
+      if (onDiskSizeWithHeader == -1) {
+        // The caller does not know the block size. Need to get it from the header. If header was
         // not cached (see getCachedHeader above), need to seek to pull it in. This is costly
         // and should happen very rarely. Currently happens on open of a hfile reader where we
         // read the trailer blocks to pull in the indices. Otherwise, we are reading block sizes
@@ -1697,8 +1722,27 @@ protected HFileBlock readBlockDataInternal(FSDataInputStream is, long offset,
           headerBuf = HEAP.allocate(hdrSize);
           readAtOffset(is, headerBuf, hdrSize, false, offset, pread);
           headerBuf.rewind();
+
+          // The caller didn't provide an anticipated block size and headerBuf was null, this is
+          // probably the first time this HDFS block has been read. The value we just read has not
+          // had HBase checksum validation ; assume it was not protected by HDFS checksum either.
+          // Sanity check the value. If it doesn't seem right, either trigger fall-back to hdfs
+          // checksum or abort the read.
+          //
+          // TODO: Should we also check the value vs. some multiple of fileContext.getBlocksize() ?
+          onDiskSizeWithHeader = getOnDiskSizeWithHeader(headerBuf, checksumSupport);
+          if (!checkOnDiskSizeWithHeader(onDiskSizeWithHeader)) {
+            if (verifyChecksum) {
+              invalidateNextBlockHeader();
+              span.addEvent("Falling back to HDFS checksumming.", attributesBuilder.build());
+              return null;
+            } else {
+              throw new IOException("Read invalid onDiskSizeWithHeader=" + onDiskSizeWithHeader);
+            }
+          }
+        } else {
+          onDiskSizeWithHeader = getOnDiskSizeWithHeader(headerBuf, checksumSupport);
         }
-        onDiskSizeWithHeader = getOnDiskSizeWithHeader(headerBuf, checksumSupport);
       }
       int preReadHeaderSize = headerBuf == null ? 0 : hdrSize;
       // Allocate enough space to fit the next block's header too; saves a seek next time through.
@@ -1721,14 +1765,36 @@ protected HFileBlock readBlockDataInternal(FSDataInputStream is, long offset,
         if (headerBuf == null) {
           headerBuf = onDiskBlock.duplicate().position(0).limit(hdrSize);
         }
-        // Do a few checks before we go instantiate HFileBlock.
-        assert onDiskSizeWithHeader > this.hdrSize;
-        verifyOnDiskSizeMatchesHeader(onDiskSizeWithHeader, headerBuf, offset, checksumSupport);
+
         ByteBuff curBlock = onDiskBlock.duplicate().position(0).limit(onDiskSizeWithHeader);
         // Verify checksum of the data before using it for building HFileBlock.
         if (verifyChecksum && !validateChecksum(offset, curBlock, hdrSize)) {
+          invalidateNextBlockHeader();
+          span.addEvent("Falling back to HDFS checksumming.", attributesBuilder.build());
           return null;
         }
+
+        // TODO: is this check necessary or can we proceed with a provided value regardless of
+        // what is in the header?
+        int fromHeader = getOnDiskSizeWithHeader(headerBuf, checksumSupport);
+        if (onDiskSizeWithHeader != fromHeader) {
+          if (LOG.isTraceEnabled()) {
+            LOG.trace("Passed in onDiskSizeWithHeader={} != {}, offset={}, fileContext={}",
+              onDiskSizeWithHeader, fromHeader, offset, this.fileContext);
+          }
+          if (checksumSupport && verifyChecksum) {
+            // This file supports HBase checksums and verification of those checksums was
+            // requested. The block size provided by the caller (presumably from the block index)
+            // does not match the block size written to the block header. treat this as
+            // HBase-checksum failure.
+            span.addEvent("Falling back to HDFS checksumming.", attributesBuilder.build());
+            invalidateNextBlockHeader();
+            return null;
+          }
+          throw new IOException("Passed in onDiskSizeWithHeader=" + onDiskSizeWithHeader + " != "
+            + fromHeader + ", offset=" + offset + ", fileContext=" + this.fileContext);
+        }
+
         // remove checksum from buffer now that it's verified
         int sizeWithoutChecksum = curBlock.getInt(Header.ON_DISK_DATA_SIZE_WITH_HEADER_INDEX);
         curBlock.limit(sizeWithoutChecksum);

hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestChecksum.java

@@ -61,7 +61,7 @@ public class TestChecksum {
   public static final HBaseClassTestRule CLASS_RULE =
     HBaseClassTestRule.forClass(TestChecksum.class);
 
-  private static final Logger LOG = LoggerFactory.getLogger(TestHFileBlock.class);
+  private static final Logger LOG = LoggerFactory.getLogger(TestChecksum.class);
 
   static final Compression.Algorithm[] COMPRESSION_ALGORITHMS = { NONE, GZ };