[HUDI-2780] Fix the issue of Mor log skipping complete blocks when re…

…ading data

hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFileReader.java

@@ -179,6 +179,7 @@ private HoodieLogBlock readBlock() throws IOException {
 
     int blocksize;
     int type;
+    long blockStartPos = inputStream.getPos();
     HoodieLogBlockType blockType = null;
     Map<HeaderMetadataType, String> header = null;
 
@@ -194,7 +195,7 @@ private HoodieLogBlock readBlock() throws IOException {
     // We may have had a crash which could have written this block partially
     // Skip blocksize in the stream and we should either find a sync marker (start of the next
     // block) or EOF. If we did not find either of it, then this block is a corrupted block.
-    boolean isCorrupted = isBlockCorrupt(blocksize);
+    boolean isCorrupted = isBlockCorrupt(blockStartPos, blocksize);
     if (isCorrupted) {
       return createCorruptBlock();
     }
@@ -281,7 +282,7 @@ private HoodieLogBlock createCorruptBlock() throws IOException {
         contentPosition, corruptedBlockSize, corruptedBlockSize, new HashMap<>(), new HashMap<>());
   }
 
-  private boolean isBlockCorrupt(int blocksize) throws IOException {
+  private boolean isBlockCorrupt(long blockStartPos, int blocksize) throws IOException {
     long currentPos = inputStream.getPos();
     try {
       inputStream.seek(currentPos + blocksize);
@@ -291,7 +292,7 @@ private boolean isBlockCorrupt(int blocksize) throws IOException {
       // This seek is required because contract of seek() is different for naked DFSInputStream vs BufferedFSInputStream
       // release-3.1.0-RC1/DFSInputStream.java#L1455
       // release-3.1.0-RC1/BufferedFSInputStream.java#L73
-      inputStream.seek(currentPos);
+      inputStream.seek(blockStartPos);
       return true;
     }