static long averageBytesPerRecord(HoodieTimeline commitTimeline, HoodieWriteConfig hoodieWriteConfig) {
    long avgSize = hoodieWriteConfig.getCopyOnWriteRecordSizeEstimate();
    long fileSizeThreshold = (long) (hoodieWriteConfig.getRecordSizeEstimationThreshold() * hoodieWriteConfig.getParquetSmallFileLimit());
    if (!commitTimeline.empty()) {
      // Go over the reverse ordered commits to get a more recent estimate of average record size.
      Iterator<HoodieInstant> instants = commitTimeline.getReverseOrderedInstants().iterator();
      while (instants.hasNext()) {
        HoodieInstant instant = instants.next();
        try {
          HoodieCommitMetadata commitMetadata = HoodieCommitMetadata
              .fromBytes(commitTimeline.getInstantDetails(instant).get(), HoodieCommitMetadata.class);
          if (instant.getAction().equals(COMMIT_ACTION) || instant.getAction().equals(REPLACE_COMMIT_ACTION)) {
            long totalBytesWritten = commitMetadata.fetchTotalBytesWritten();
            long totalRecordsWritten = commitMetadata.fetchTotalRecordsWritten();
            if (totalBytesWritten > fileSizeThreshold && totalRecordsWritten > 0) {
              avgSize = (long) Math.ceil((1.0 * totalBytesWritten) / totalRecordsWritten);
              break;
            }
          } else if (instant.getAction().equals(DELTA_COMMIT_ACTION)) {
            // lets consider only base files in case of delta commits
            AtomicLong totalBytesWritten = new AtomicLong(0L);
            AtomicLong totalRecordsWritten = new AtomicLong(0L);
            commitMetadata.getWriteStats().stream()
                .filter(hoodieWriteStat -> FSUtils.isBaseFile(new StoragePath(hoodieWriteStat.getPath())))
                .forEach(hoodieWriteStat -> {
                  totalBytesWritten.addAndGet(hoodieWriteStat.getTotalWriteBytes());
                  totalRecordsWritten.addAndGet(hoodieWriteStat.getNumWrites());
                });
            if (totalBytesWritten.get() > fileSizeThreshold && totalRecordsWritten.get() > 0) {
              avgSize = (long) Math.ceil((1.0 * totalBytesWritten.get()) / totalRecordsWritten.get());
              break;
            }
          }
        } catch (Throwable t) {
          // make this fail safe.
          LOG.error("Error trying to compute average bytes/record ", t);
        }
      }
    }
    return avgSize;
  }
@Test
  public void testErrorHandling() {
    int recordSize = 10000;
    HoodieWriteConfig writeConfig = HoodieWriteConfig.newBuilder()
        .withProps(Collections.singletonMap(COPY_ON_WRITE_RECORD_SIZE_ESTIMATE.key(), String.valueOf(recordSize)))
        .build(false);
    HoodieDefaultTimeline commitsTimeline = new HoodieDefaultTimeline();
    List<HoodieInstant> instants = Collections.singletonList(
        new HoodieInstant(HoodieInstant.State.COMPLETED, HoodieTimeline.COMMIT_ACTION, "1"));

    when(mockTimeline.getInstants()).thenReturn(instants);
    when(mockTimeline.getReverseOrderedInstants()).then(i -> instants.stream());
    // Simulate a case where the instant details are absent
    commitsTimeline.setInstants(new ArrayList<>());

    assertEquals(recordSize, AverageRecordSizeUtils.averageBytesPerRecord(mockTimeline, writeConfig));
  }