public void switchToUpsert() throws Exception {
        if (!taskWriterFactory.isUpsert()) {
            LOGGER.info("iceberg writer switch to upsert write mode");
            taskWriterFactory.switchToUpsert();
            cacheWriteResultAndRecreateWriter();
        }
    }
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