@Override
    public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
        // submit the cached write results
        LOGGER.info("Submit {} cached write results before checkpoint {}.",
                cachedWriteResults.size(), checkpointId);
        cachedWriteResults.forEach(this::emit);
        cachedWriteResults.clear();
        // close all open files and emit files to downstream committer operator
        emit(writer.complete());
        this.writer = taskWriterFactory.create();
    }
this method should only be called in open() method
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/IcebergSingleStreamWriter.java