Fix bug with async direct IO fetching that caused misreading of bytes (#136213)

Author: benwtrent

server/src/main/java/org/elasticsearch/index/store/AsyncDirectIOIndexInput.java

@@ -523,6 +523,7 @@ boolean readBytes(long pos, ByteBuffer slice, int delta) throws IOException {
         }
 
         void clearSlotAndMaybeStartPending(int slot) {
+            assert prefetchThreads.get(slot) != null && prefetchThreads.get(slot).isDone();
             prefetchThreads.set(slot, null);
             posToSlot.remove(prefetchPos[slot]);
             if (pendingPrefetches.isEmpty()) {
@@ -531,9 +532,22 @@ void clearSlotAndMaybeStartPending(int slot) {
             }
             final long req = pendingPrefetches.removeFirst();
             posToSlot.put(req, slot);
+            prefetchPos[slot] = req;
             startPrefetch(req, slot);
         }
 
+        private boolean assertSlotsConsistent() {
+            posToSlot.forEach((k, v) -> {
+                if (prefetchThreads.get(v) == null) {
+                    throw new AssertionError("posToSlot inconsistent: slot " + v + " for pos " + k + " has no prefetch thread");
+                }
+                if (prefetchPos[v] != k) {
+                    throw new AssertionError("posToSlot inconsistent: slot " + v + " for pos " + k + " has prefetchPos " + prefetchPos[v]);
+                }
+            });
+            return true;
+        }
+
         void startPrefetch(long pos, int slot) {
             Future<ByteBuffer> future = executor.submit(() -> {
                 var prefetchBuffers = DirectIOPrefetcher.this.prefetchBuffers;
@@ -550,6 +564,7 @@ void startPrefetch(long pos, int slot) {
                 return prefetchBuffer;
             });
             prefetchThreads.set(slot, future);
+            assert assertSlotsConsistent();
         }
 
         @Override

server/src/test/java/org/elasticsearch/index/store/AsyncDirectIODirectoryTests.java

@@ -44,6 +44,7 @@
 import java.nio.file.Path;
 
 import static org.apache.lucene.store.FSDirectory.open;
+import static org.elasticsearch.test.ESTestCase.randomIntBetween;
 
 public class AsyncDirectIODirectoryTests extends BaseDirectoryTestCase {
 
@@ -74,7 +75,7 @@ private static int getBlockSize(Path path) throws IOException {
 
     @Override
     protected Directory getDirectory(Path path) throws IOException {
-        return new FsDirectoryFactory.AlwaysDirectIODirectory(open(path), 8192, 8192, 32);
+        return new FsDirectoryFactory.AlwaysDirectIODirectory(open(path), 8192, 8192, randomIntBetween(0, 32));
     }
 
     public void testIndexWriteRead() throws IOException {