Fix directIO reader hang when pread returns short reads

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/directentrylogger/DirectReader.java

@@ -157,7 +157,7 @@ private int readBytesIntoBuf(ByteBuf buf, long offset, int size) throws IOExcept
                                       .kv("offset", offset)
                                       .kv("size", size).toString());
             }
-            return nativeBuffer.readByteBuf(buf, offsetInBuffer, size);
+            return nativeBuffer.readByteBuf(buf, offsetInBuffer, sizeInBuffer);
         }
     }
 
@@ -207,6 +207,9 @@ void readBlock(long offset) throws IOException {
         long bytesToRead = Math.min(blockSize, bytesAvailable);
         long bytesOutstanding = bytesToRead;
         long bytesRead = -1;
+        // A failed read may still overwrite nativeBuffer, so invalidate the
+        // cached block before loading a new one.
+        clearCache();
         try {
             while (true) {
                 long readSize = blockSize - bufferOffset;
@@ -226,8 +229,21 @@ void readBlock(long offset) throws IOException {
                 if ((bytesOutstanding - bytesRead) <= 0) {
                     break;
                 }
-                bytesOutstanding -= bytesRead & Buffer.ALIGNMENT;
-                bufferOffset += bytesRead & Buffer.ALIGNMENT;
+                long alignedBytesRead = bytesRead & ~(Buffer.ALIGNMENT - 1L);
+                if (alignedBytesRead <= 0) {
+                    readBlockStats.registerFailedEvent(System.nanoTime() - startNs, TimeUnit.NANOSECONDS);
+                    throw new IOException(exMsg("Short read did not make aligned progress")
+                                          .kv("requestedBytes", readSize)
+                                          .kv("offset", blockStart + bufferOffset)
+                                          .kv("expectedBytes", Math.min(blockSize, bytesAvailable))
+                                          .kv("bytesOutstanding", bytesOutstanding)
+                                          .kv("bufferOffset", bufferOffset)
+                                          .kv("bytesRead", bytesRead)
+                                          .kv("file", filename)
+                                          .kv("fd", fd).toString());
+                }
+                bytesOutstanding -= alignedBytesRead;
+                bufferOffset += alignedBytesRead;
             }
         } catch (NativeIOException ne) {
             readBlockStats.registerFailedEvent(System.nanoTime() - startNs, TimeUnit.NANOSECONDS);

bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/directentrylogger/TestDirectReader.java

@@ -392,8 +392,7 @@ public void testPartialRead() throws Exception {
         NativeIOImpl nativeIO = new NativeIOImpl() {
                 @Override
                 public long pread(int fd, long buf, long size, long offset) throws NativeIOException {
-                    long read = super.pread(fd, buf, size, offset);
-                    return Math.min(read, Buffer.ALIGNMENT); // force only less than a buffer read
+                    return super.pread(fd, buf, Math.min(size, Buffer.ALIGNMENT), offset);
                 }
 
                 @Override
@@ -445,6 +444,85 @@ buffers, new NativeIOImpl(), Logger.get(TestDirectReader.class))) {
         }
     }
 
+    @Test
+    public void testPartialReadProgressesByAlignedBytes() throws Exception {
+        File ledgerDir = tmpDirs.createNew("partialReadAligned", "logs");
+
+        writeFileWithPattern(ledgerDir, 1234, 0xbeefcafe, 1, 1 << 20);
+
+        class ShortReadNativeIO extends NativeIOImpl {
+            int calls;
+
+            @Override
+            public long pread(int fd, long buf, long size, long offset) throws NativeIOException {
+                if (calls == 0) {
+                    assertThat(offset, equalTo(0L));
+                } else if (calls == 1) {
+                    assertThat(offset, equalTo((long) Buffer.ALIGNMENT * 2));
+                }
+                calls++;
+
+                return super.pread(fd, buf, Math.min(size, Buffer.ALIGNMENT * 2L), offset);
+            }
+        }
+
+        ShortReadNativeIO nativeIO = new ShortReadNativeIO();
+        try (LogReader reader = new DirectReader(1234, logFilename(ledgerDir, 1234),
+                                                 ByteBufAllocator.DEFAULT,
+                                                 nativeIO, Buffer.ALIGNMENT * 4,
+                                                 1 << 20, opLogger)) {
+            ByteBuf bb = reader.readBufferAt(0, Buffer.ALIGNMENT * 4);
+            try {
+                for (int block = 0; block < 4; block++) {
+                    for (int i = 0; i < Buffer.ALIGNMENT / Integer.BYTES; i++) {
+                        assertThat(bb.readInt(), equalTo(0xbeefcafe + block));
+                    }
+                }
+                assertThat(bb.readableBytes(), equalTo(0));
+            } finally {
+                bb.release();
+            }
+        }
+        assertThat(nativeIO.calls, equalTo(2));
+    }
+
+    @Test
+    public void testFailedPartialReadInvalidatesCachedBlock() throws Exception {
+        File ledgerDir = tmpDirs.createNew("partialReadCache", "logs");
+
+        writeFileWithPattern(ledgerDir, 1234, 0xbeefcafe, 1, 1 << 20);
+
+        class FailingShortReadNativeIO extends NativeIOImpl {
+            int calls;
+
+            @Override
+            public long pread(int fd, long buf, long size, long offset) throws NativeIOException {
+                calls++;
+                if (calls == 2) {
+                    return super.pread(fd, buf, Buffer.ALIGNMENT * 2L, offset);
+                } else if (calls == 3) {
+                    return 0;
+                }
+                return super.pread(fd, buf, size, offset);
+            }
+        }
+
+        FailingShortReadNativeIO nativeIO = new FailingShortReadNativeIO();
+        try (LogReader reader = new DirectReader(1234, logFilename(ledgerDir, 1234),
+                                                 ByteBufAllocator.DEFAULT,
+                                                 nativeIO, Buffer.ALIGNMENT * 4,
+                                                 1 << 20, opLogger)) {
+            assertThat(reader.readIntAt(0), equalTo(0xbeefcafe));
+
+            IOException exception = Assertions.assertThrows(
+                    IOException.class, () -> reader.readIntAt(Buffer.ALIGNMENT * 4L));
+            Assertions.assertFalse(exception instanceof EOFException);
+
+            assertThat(reader.readIntAt(0), equalTo(0xbeefcafe));
+        }
+        assertThat(nativeIO.calls, equalTo(4));
+    }
+
     @Test
     public void testLargeEntry() throws Exception {
         File ledgerDir = tmpDirs.createNew("largeEntries", "logs");