void updateFlushResult(WriteResult result) {
        flushedDataFiles.inc(result.dataFiles().length);
        flushedDeleteFiles.inc(result.deleteFiles().length);
        flushedReferencedDataFiles.inc(result.referencedDataFiles().length);

        // For file size distribution histogram, we don't have to update them after successful commits.
        // This should works equally well and we avoided the overhead of tracking the list of file sizes
        // in the {@link CommitSummary}, which currently stores simple stats for counters and gauges
        // metrics.
        Arrays.stream(result.dataFiles())
                .forEach(
                        dataFile -> {
                            dataFilesSizeHistogram.update(dataFile.fileSizeInBytes());
                        });
        Arrays.stream(result.deleteFiles())
                .forEach(
                        deleteFile -> {
                            deleteFilesSizeHistogram.update(deleteFile.fileSizeInBytes());
                        });
    }
Copy from iceberg-flink:iceberg-flink-1.15:1.3.1
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/IcebergStreamWriterMetrics.java