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);
         }
     }
 
@@ -226,8 +226,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 EOFException(exMsg("Short read did not make aligned progress")
+                                          .kv("requestedBytes", blockSize)
+                                          .kv("offset", blockStart)
+                                          .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

@@ -445,6 +445,49 @@ 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++;
+
+                long read = super.pread(fd, buf, size, offset);
+                return Math.min(read, Buffer.ALIGNMENT * 2L);
+            }
+        }
+
+        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 testLargeEntry() throws Exception {
         File ledgerDir = tmpDirs.createNew("largeEntries", "logs");