Centralize common code for footer checksum read in BaseSearchableSnapshotIndexInput class (#68902) (#69015)

This commit factorizes in the BaseSearchableSnapshotIndexInput
class some common code used to read the footer checksum of
searchable snapshots IndexInput implementations.

This change helps to maintain the code to read the footer checksum
in one place and helps to enable this optimisation for
DirectBlobContainerIndexInput. It also helps to assert the non-usage
of the cache fetch async thread pool when reading from a index input
in a single place.

Finally, the index input unit tests have been adapted to account for
the footer checksum read optimisation: they now generate a binary
content that contains a valid footer checksum.

Backport of #68902

Author: tlrx

test/framework/src/main/java/org/elasticsearch/test/ESTestCase.java

@@ -668,6 +668,11 @@ public static byte randomByte() {
         return (byte) random().nextInt();
     }
 
+    public static byte randomNonNegativeByte() {
+        byte randomByte =  randomByte();
+        return (byte) (randomByte == Byte.MIN_VALUE ? 0 : Math.abs(randomByte));
+    }
+
     /**
      * Helper method to create a byte array of a given length populated with random byte values
      *

x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/index/store/BaseSearchableSnapshotIndexInput.java

@@ -6,6 +6,8 @@
  */
 package org.elasticsearch.index.store;
 
+import org.apache.logging.log4j.Logger;
+import org.apache.lucene.codecs.CodecUtil;
 import org.apache.lucene.store.BufferedIndexInput;
 import org.apache.lucene.store.IOContext;
 import org.elasticsearch.common.blobstore.BlobContainer;
@@ -16,11 +18,16 @@
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.nio.ByteBuffer;
 import java.util.Objects;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Predicate;
+
+import static org.elasticsearch.index.store.checksum.ChecksumBlobContainerIndexInput.checksumToBytesArray;
 
 public abstract class BaseSearchableSnapshotIndexInput extends BufferedIndexInput {
 
+    protected final Logger logger;
     protected final BlobContainer blobContainer;
     protected final FileInfo fileInfo;
     protected final IOContext context;
@@ -33,6 +40,7 @@ public abstract class BaseSearchableSnapshotIndexInput extends BufferedIndexInpu
     private AtomicBoolean closed;
 
     public BaseSearchableSnapshotIndexInput(
+        Logger logger,
         String resourceDesc,
         BlobContainer blobContainer,
         FileInfo fileInfo,
@@ -42,6 +50,7 @@ public BaseSearchableSnapshotIndexInput(
         long length
     ) {
         super(resourceDesc, context);
+        this.logger = Objects.requireNonNull(logger);
         this.blobContainer = Objects.requireNonNull(blobContainer);
         this.fileInfo = Objects.requireNonNull(fileInfo);
         this.context = Objects.requireNonNull(context);
@@ -54,25 +63,60 @@ public BaseSearchableSnapshotIndexInput(
         this.isClone = false;
     }
 
-    public BaseSearchableSnapshotIndexInput(
-        String resourceDesc,
-        BlobContainer blobContainer,
-        FileInfo fileInfo,
-        IOContext context,
-        IndexInputStats stats,
-        long offset,
-        long length,
-        int bufferSize
-    ) {
-        this(resourceDesc, blobContainer, fileInfo, context, stats, offset, length);
-        setBufferSize(bufferSize);
-    }
-
     @Override
     public final long length() {
         return length;
     }
 
+    @Override
+    protected final void readInternal(ByteBuffer b) throws IOException {
+        assert assertCurrentThreadIsNotCacheFetchAsync();
+
+        // We can detect that we're going to read the last 16 bytes (that contains the footer checksum) of the file. Such reads are often
+        // executed when opening a Directory and since we have the checksum in the snapshot metadata we can use it to fill the ByteBuffer.
+        if (maybeReadChecksumFromFileInfo(b)) {
+            logger.trace("read footer of file [{}], bypassing all caches", fileInfo.physicalName());
+            assert b.remaining() == 0L : b.remaining();
+            return;
+        }
+
+        doReadInternal(b);
+    }
+
+    protected abstract void doReadInternal(ByteBuffer b) throws IOException;
+
+    /**
+     * Detects read operations that are executed on the last 16 bytes of the index input which is where Lucene stores the footer checksum
+     * of Lucene files. If such a read is detected this method tries to complete the read operation by reading the checksum from the
+     * {@link FileInfo} in memory rather than reading the bytes from the {@link BufferedIndexInput} because that could trigger more cache
+     * operations.
+     *
+     * @return true if the footer checksum has been read from the {@link FileInfo}
+     */
+    private boolean maybeReadChecksumFromFileInfo(ByteBuffer b) throws IOException {
+        final int remaining = b.remaining();
+        if (remaining != CodecUtil.footerLength()) {
+            return false;
+        }
+        final long position = getFilePointer() + this.offset;
+        if (position != fileInfo.length() - CodecUtil.footerLength()) {
+            return false;
+        }
+        if (isClone) {
+            return false;
+        }
+        boolean success = false;
+        try {
+            b.put(checksumToBytesArray(fileInfo.checksum()));
+            success = true;
+        } catch (NumberFormatException e) {
+            // tests disable this optimisation by passing an invalid checksum
+        } finally {
+            assert b.remaining() == (success ? 0L : remaining) : b.remaining() + " remaining bytes but success is " + success;
+        }
+        return success;
+    }
+
     /**
      * Opens an {@link InputStream} for the given range of bytes which reads the data directly from the blob store. If the requested range
      * spans multiple blobs then this stream will request them in turn.
@@ -173,11 +217,18 @@ public final void close() throws IOException {
             if (isClone == false) {
                 stats.incrementCloseCount();
             }
-            innerClose();
+            doClose();
         }
     }
 
-    public abstract void innerClose() throws IOException;
+    public abstract void doClose() throws IOException;
+
+    protected void ensureContext(Predicate<IOContext> predicate) throws IOException {
+        if (predicate.test(context) == false) {
+            assert false : "this method should not be used with this context " + context;
+            throw new IOException("Cannot read the index input using context [context=" + context + ", input=" + this + ']');
+        }
+    }
 
     protected final boolean assertCurrentThreadMayAccessBlobStore() {
         final String threadName = Thread.currentThread().getName();
@@ -199,4 +250,15 @@ protected final boolean assertCurrentThreadMayAccessBlobStore() {
         return true;
     }
 
+    protected static boolean isCacheFetchAsyncThread(final String threadName) {
+        return threadName.contains('[' + SearchableSnapshotsConstants.CACHE_FETCH_ASYNC_THREAD_POOL_NAME + ']');
+    }
+
+    protected static boolean assertCurrentThreadIsNotCacheFetchAsync() {
+        final String threadName = Thread.currentThread().getName();
+        assert false == isCacheFetchAsyncThread(threadName) : "expected the current thread ["
+            + threadName
+            + "] to belong to the cache fetch async thread pool";
+        return true;
+    }
 }

x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/index/store/cache/CachedBlobContainerIndexInput.java

@@ -10,7 +10,6 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
-import org.apache.lucene.codecs.CodecUtil;
 import org.apache.lucene.store.AlreadyClosedException;
 import org.apache.lucene.store.IOContext;
 import org.apache.lucene.store.IndexInput;
@@ -29,7 +28,6 @@
 import org.elasticsearch.index.store.BaseSearchableSnapshotIndexInput;
 import org.elasticsearch.index.store.IndexInputStats;
 import org.elasticsearch.index.store.SearchableSnapshotDirectory;
-import org.elasticsearch.xpack.searchablesnapshots.SearchableSnapshotsConstants;
 import org.elasticsearch.xpack.searchablesnapshots.cache.ByteRange;
 
 import java.io.EOFException;
@@ -42,10 +40,8 @@
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
-import java.util.function.Predicate;
 import java.util.stream.IntStream;
 
-import static org.elasticsearch.index.store.checksum.ChecksumBlobContainerIndexInput.checksumToBytesArray;
 import static org.elasticsearch.xpack.searchablesnapshots.SearchableSnapshotsUtils.toIntBytes;
 
 public class CachedBlobContainerIndexInput extends BaseSearchableSnapshotIndexInput {
@@ -106,7 +102,7 @@ private CachedBlobContainerIndexInput(
         int rangeSize,
         int recoveryRangeSize
     ) {
-        super(resourceDesc, directory.blobContainer(), fileInfo, context, stats, offset, length);
+        super(logger, resourceDesc, directory.blobContainer(), fileInfo, context, stats, offset, length);
         this.directory = directory;
         this.cacheFileReference = cacheFileReference;
         this.lastReadPosition = this.offset;
@@ -116,19 +112,12 @@ private CachedBlobContainerIndexInput(
     }
 
     @Override
-    public void innerClose() {
+    public void doClose() {
         if (isClone == false) {
             cacheFileReference.releaseOnClose();
         }
     }
 
-    private void ensureContext(Predicate<IOContext> predicate) throws IOException {
-        if (predicate.test(context) == false) {
-            assert false : "this method should not be used with this context " + context;
-            throw new IOException("Cannot read the index input using context [context=" + context + ", input=" + this + ']');
-        }
-    }
-
     private long getDefaultRangeSize() {
         return (context != CACHE_WARMING_CONTEXT)
             ? (directory.isRecoveryFinalized() ? defaultRangeSize : recoveryRangeSize)
@@ -143,24 +132,12 @@ private ByteRange computeRange(long position) {
     }
 
     @Override
-    protected void readInternal(ByteBuffer b) throws IOException {
+    protected void doReadInternal(ByteBuffer b) throws IOException {
         ensureContext(ctx -> ctx != CACHE_WARMING_CONTEXT);
-        assert assertCurrentThreadIsNotCacheFetchAsync();
         final long position = getFilePointer() + this.offset;
         final int length = b.remaining();
 
-        // We can detect that we're going to read the last 16 bytes (that contains the footer checksum) of the file. Such reads are often
-        // executed when opening a Directory and since we have the checksum in the snapshot metadata we can use it to fill the ByteBuffer.
-        if (length == CodecUtil.footerLength() && isClone == false && position == fileInfo.length() - length) {
-            if (readChecksumFromFileInfo(b)) {
-                logger.trace("read footer of file [{}] at position [{}], bypassing all caches", fileInfo.physicalName(), position);
-                return;
-            }
-            assert b.remaining() == length;
-        }
-
         logger.trace("readInternal: read [{}-{}] ([{}] bytes) from [{}]", position, position + length, length, this);
-
         try {
             final CacheFile cacheFile = cacheFileReference.get();
 
@@ -401,26 +378,6 @@ private int readDirectlyIfAlreadyClosed(long position, ByteBuffer b, Exception e
         throw new IOException("failed to read data from cache", e);
     }
 
-    private boolean readChecksumFromFileInfo(ByteBuffer b) throws IOException {
-        assert isClone == false;
-        byte[] footer;
-        try {
-            footer = checksumToBytesArray(fileInfo.checksum());
-        } catch (NumberFormatException e) {
-            // tests disable this optimisation by passing an invalid checksum
-            footer = null;
-        }
-        if (footer == null) {
-            return false;
-        }
-
-        b.put(footer);
-        assert b.remaining() == 0L;
-        return true;
-
-        // TODO we should add this to DirectBlobContainerIndexInput too.
-    }
-
     /**
      * Prefetches a complete part and writes it in cache. This method is used to prewarm the cache.
      * @return a tuple with {@code Tuple<Persistent Cache Length, Prefetched Length>} values
@@ -737,23 +694,11 @@ private static boolean assertFileChannelOpen(FileChannel fileChannel) {
         return true;
     }
 
-    private static boolean isCacheFetchAsyncThread(final String threadName) {
-        return threadName.contains('[' + SearchableSnapshotsConstants.CACHE_FETCH_ASYNC_THREAD_POOL_NAME + ']');
-    }
-
     private static boolean assertCurrentThreadMayWriteCacheFile() {
         final String threadName = Thread.currentThread().getName();
         assert isCacheFetchAsyncThread(threadName) : "expected the current thread ["
             + threadName
             + "] to belong to the cache fetch async thread pool";
         return true;
     }
-
-    private static boolean assertCurrentThreadIsNotCacheFetchAsync() {
-        final String threadName = Thread.currentThread().getName();
-        assert false == isCacheFetchAsyncThread(threadName) : "expected the current thread ["
-            + threadName
-            + "] to belong to the cache fetch async thread pool";
-        return true;
-    }
 }