@Override
    public void dispose() throws Exception {
        if (writer != null) {
            writer.close();
            writer = null;
        }
    }
check if the data is incremental phase
by checking the incremental field
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/IcebergSingleStreamWriter.java