@Override
    public PendingFileRecoverable closeForCommit() throws IOException {
        if (sinkMultipleEnable) {
            LOG.info("record writer cache {}", CacheHolder.getRecordWriterHashMap());
            Iterator<Path> iterator = CacheHolder.getRecordWriterHashMap().keySet().iterator();
            while (iterator.hasNext()) {
                Path inProgressFilePath = iterator.next();
                // one flink batch writes many hive tables, they are the same inProgressPath
                if (inProgressFilePath.getName().equals(this.inProgressPath.getName())) {
                    FileSinkOperator.RecordWriter recordWriter = CacheHolder.getRecordWriterHashMap()
                            .get(inProgressFilePath);
                    writer.setRecordWriter(recordWriter);
                    writer.flush();
                    writer.finish();
                    // clear cache
                    iterator.remove();
                    CacheHolder.getRowConverterHashMap().remove(inProgressFilePath);

                    // parse the target location of hive table
                    String tmpFileName = inProgressFilePath.getName();
                    String targetPathName = tmpFileName.substring(1, tmpFileName.lastIndexOf(".inprogress"));
                    Path targetPath = new Path(inProgressFilePath.getParent() + "/" + targetPathName);

                    LOG.info("file committer target path {}, in progress file {}", targetPath, inProgressFilePath);
                    HadoopRenameFileCommitter committer = new HadoopRenameFileCommitter(
                            ((HadoopRenameFileCommitter) fileCommitter).getConfiguration(),
                            targetPath,
                            inProgressFilePath,
                            true);
                    CacheHolder.getFileCommitterHashMap().put(inProgressFilePath, committer);
                }
            }
        } else {
            writer.flush();
            writer.finish();
            fileCommitter.preCommit();
        }
        return new HadoopPathBasedPendingFile(fileCommitter, getSize()).getRecoverable();
    }
check if source table schema changes

@param identifier hive database name and table name
@param writerFactory hive writer factory
@param schema hive field with flink types
@return if schema has changed
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/hive/src/main/java/org/apache/inlong/sort/hive/filesystem/HadoopPathBasedPartFileWriter.java