[Bug][Sort] Table not found when Iceberg commit Hive table

[thesumery]

What happened

What you expected to happen

succesful to commit hive table

How to reproduce

1. prepare an environment：inlong 1.3.0-snapshot、flink 1.13.5、hive metastore 2.3.7、hive server 2.3.7、mysql 5.7
2. In Mysql, create a table and open binlog

create table if not exists `stu` (
    `age` INT,
    `id` INT,
    `name` VARCHAR(63),
    PRIMARY KEY(id)
);

4. In Hive server, beeline to connect it and create a iceberg table

 CREATE EXTERNAL TABLE `stu_ice`(                   
   `id` int COMMENT 'from deserializer',            
   `name` string COMMENT 'from deserializer',       
   `age` int COMMENT 'from deserializer')           
 ROW FORMAT SERDE                                   
   'org.apache.iceberg.mr.hive.HiveIcebergSerDe'    
 STORED BY                                          
   'org.apache.iceberg.mr.hive.HiveIcebergStorageHandler'                                    
 LOCATION                                           
   'hdfs://loaclhost:4007/usr/hive/warehouse/cran_test.db/stu' 
 TBLPROPERTIES (                                    
   'engine.hive.enabled'='true',                    
   'external.table.purge'='TRUE',                   
   'format-version'='2',                            
   'property-version'='2',                          
   'table_type'='ICEBERG')

5. In Flink sql client，in memory catalog create a mysql table and iceberg table, load data into iceberg from mysql.

CREATE TABLE `table_913053` (
    `age` INT,
    `id` INT,
    `name` STRING,
    `PROCESS_TIME` AS PROCTIME()
) WITH (
    'scan.incremental.snapshot.enabled' = 'false',
    'connector' = 'mysql-cdc-inlong',
    'hostname' = 'localhost',
    'username' = 'inlong',
    'password' = 'inlong',
    'database-name' = 'test',
    'port' = '3306',
    'server-time-zone' = 'Asia/Shanghai',
    'table-name' = 'stu'
);
CREATE TABLE `stu_ice`(
    PRIMARY KEY(`id`),
    `id` INT,
    `name` STRING,
    `age` INT)
WITH (
    'connector' = 'iceberg',
    'catalog-database' = 'test',
    'catalog-table' = 'stu_ice',
    'catalog-type' = 'HIVE',
    'catalog-name' = 'HIVE',
    'uri' = 'thrift://localhost:7004',  -- hive metastore uri
    'warehouse' = '/usr/hive/warehouse'
);
insert into `stu` 
    select `id`, `age`, `name`
    from `table_913053`;

Environment

CentOS Linux release 7.2

InLong version

master

InLong Component

InLong Sort

Are you willing to submit PR?

• Yes, I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[github-actions]

Thanks a lot for opening your first issue with us! 🧡 We'll get back to you shortly! ⏳