public boolean canWrite() {
    long writtenCount = getWrittenRecordCount();
    if (writtenCount >= recordCountForNextSizeCheck) {
      long dataSize = getDataSize();
      // In some very extreme cases, like all records are same value, then it's possible
      // the dataSize is much lower than the writtenRecordCount(high compression ratio),
      // causing avgRecordSize to 0, we'll force the avgRecordSize to 1 for such cases.
      long avgRecordSize = Math.max(dataSize / writtenCount, 1);
      // Follow the parquet block size check logic here, return false
      // if it is within ~2 records of the limit
      if (dataSize > (maxFileSize - avgRecordSize * 2)) {
        return false;
      }
      recordCountForNextSizeCheck = writtenCount + Math.min(
          // Do check it in the halfway
          Math.max(ParquetProperties.DEFAULT_MINIMUM_RECORD_COUNT_FOR_CHECK, (maxFileSize / avgRecordSize - writtenCount) / 2),
          ParquetProperties.DEFAULT_MAXIMUM_RECORD_COUNT_FOR_CHECK);
    }
    return true;
  }
@Test
  public void testCanWrite() throws IOException {
    BloomFilter filter = BloomFilterFactory.createBloomFilter(1000, 0.0001, 10000,
        BloomFilterTypeCode.DYNAMIC_V0.name());
    StorageConfiguration conf = HoodieTestUtils.getDefaultStorageConfWithDefaults();

    Schema schema = new Schema.Parser().parse(HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA);
    HoodieAvroWriteSupport writeSupport = new HoodieAvroWriteSupport(new AvroSchemaConverter().convert(schema),
        schema, Option.of(filter), new Properties());

    long maxFileSize = 2 * 1024 * 1024;
    HoodieParquetConfig<HoodieAvroWriteSupport> parquetConfig =
        new HoodieParquetConfig<>(writeSupport, CompressionCodecName.GZIP, ParquetWriter.DEFAULT_BLOCK_SIZE,
            ParquetWriter.DEFAULT_PAGE_SIZE, maxFileSize, conf, 0, true);

    StoragePath filePath = new StoragePath(
        new StoragePath(tempDir.toUri()), "test_fileSize.parquet");
    try (MockHoodieParquetWriter writer = new MockHoodieParquetWriter(filePath, parquetConfig)) {
      // doesn't start write, should return true
      assertTrue(writer.canWrite());
      // recordCountForNextSizeCheck should be DEFAULT_MINIMUM_RECORD_COUNT_FOR_CHECK
      assertEquals(writer.getRecordCountForNextSizeCheck(), DEFAULT_MINIMUM_RECORD_COUNT_FOR_CHECK);
      // 10 bytes per record
      writer.setCurrentDataSize(1000);
      writer.setWrittenRecordCount(writer.getRecordCountForNextSizeCheck());

      assertTrue(writer.canWrite());
      // Should check it with more DEFAULT_MAXIMUM_RECORD_COUNT_FOR_CHECK records
      assertEquals(writer.getRecordCountForNextSizeCheck(), writer.getWrittenRecordCount() + DEFAULT_MAXIMUM_RECORD_COUNT_FOR_CHECK);

      // 80 bytes per record
      writer.setCurrentDataSize(808000);
      writer.setWrittenRecordCount(writer.getRecordCountForNextSizeCheck());
      assertTrue(writer.canWrite());
      // Should check it half way, not DEFAULT_MAXIMUM_RECORD_COUNT_FOR_CHECK
      long avgRecordSize = writer.getDataSize() / writer.getWrittenRecordCount();
      long recordsDelta = (maxFileSize / avgRecordSize - writer.getWrittenRecordCount()) / 2;
      assertEquals(writer.getRecordCountForNextSizeCheck(), writer.getWrittenRecordCount() + recordsDelta);

      writer.setCurrentDataSize(maxFileSize);
      writer.setWrittenRecordCount(writer.getRecordCountForNextSizeCheck());
      assertFalse(writer.canWrite(),
          "The writer stops write new records while the file doesn't reach the max file size limit");
    }
  }