chore: abort last not complete batch (#91)

Co-authored-by: 赵海源 <[email protected]>

Author: zhaohaidao

crates/fluss/src/record/arrow.rs

@@ -369,7 +369,11 @@ impl<'a> LogRecordsBatchs<'a> {
 
         let batch_size_bytes =
             LittleEndian::read_i32(self.data.get(self.current_pos + LENGTH_OFFSET..).unwrap());
-        Some(batch_size_bytes as usize + LOG_OVERHEAD)
+        let batch_size = batch_size_bytes as usize + LOG_OVERHEAD;
+        if batch_size > self.remaining_bytes {
+            return None;
+        }
+        Some(batch_size)
     }
 }