[HUDI-3709] Fixing ParquetWriter impls not respecting Parquet Max File Size limit

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/storage/HoodieFileWriter.java

@@ -37,8 +37,6 @@ public interface HoodieFileWriter<R extends IndexedRecord> {
 
   void writeAvro(String key, R oldRecord) throws IOException;
 
-  long getBytesWritten();
-
   default void prepRecordWithMetadata(R avroRecord, HoodieRecord record, String instantTime, Integer partitionId, AtomicLong recordIndex, String fileName) {
     String seqId = HoodieRecord.generateSequenceId(instantTime, partitionId, recordIndex.getAndIncrement());
     HoodieAvroUtils.addHoodieKeyToRecord((GenericRecord) avroRecord, record.getRecordKey(), record.getPartitionPath(), fileName);

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/storage/HoodieHFileWriter.java

@@ -187,9 +187,4 @@ public void readFields(DataInput in) throws IOException {
     writer.close();
     writer = null;
   }
-
-  @Override
-  public long getBytesWritten() {
-    return fs.getBytesWritten(file);
-  }
 }

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/storage/HoodieOrcWriter.java

@@ -165,9 +165,4 @@ public void close() throws IOException {
 
     writer.close();
   }
-
-  @Override
-  public long getBytesWritten() {
-    return fs.getBytesWritten(file);
-  }
 }

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/storage/HoodieParquetWriter.java

@@ -97,7 +97,7 @@ public void writeAvroWithMetadata(R avroRecord, HoodieRecord record) throws IOEx
 
   @Override
   public boolean canWrite() {
-    return fs.getBytesWritten(file) < maxFileSize;
+    return getDataSize() < maxFileSize;
   }
 
   @Override
@@ -107,9 +107,4 @@ public void writeAvro(String key, IndexedRecord object) throws IOException {
       writeSupport.add(key);
     }
   }
-
-  @Override
-  public long getBytesWritten() {
-    return fs.getBytesWritten(file);
-  }
 }

hudi-client/hudi-java-client/src/test/java/org/apache/hudi/table/action/commit/TestJavaCopyOnWriteActionExecutor.java

@@ -402,7 +402,7 @@ public void testFileSizeUpsertRecords() throws Exception {
         counts++;
       }
     }
-    assertEquals(3, counts, "If the number of records are more than 1150, then there should be a new file");
+    assertEquals(5, counts, "If the number of records are more than 1150, then there should be a new file");
   }
 
   @Test

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/row/HoodieInternalRowParquetWriter.java

@@ -56,7 +56,7 @@ public HoodieInternalRowParquetWriter(Path file, HoodieRowParquetConfig parquetC
 
   @Override
   public boolean canWrite() {
-    return fs.getBytesWritten(file) < maxFileSize;
+    return getDataSize() < maxFileSize;
   }
 
   @Override

hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/commit/TestCopyOnWriteActionExecutor.java

@@ -437,7 +437,7 @@ public void testFileSizeUpsertRecords() throws Exception {
         counts++;
       }
     }
-    assertEquals(3, counts, "If the number of records are more than 1150, then there should be a new file");
+    assertEquals(5, counts, "If the number of records are more than 1150, then there should be a new file");
   }
 
   @Test