Flakey flink unit tests TestFlinkTableSink#testHashDistributeMode

[openinx]

I encountered the flakey unit tests several times:

org.apache.iceberg.flink.TestFlinkTableSink > testHashDistributeMode[catalogName=testhive, baseNamespace=, format=AVRO, isStreaming=true] FAILED
    java.lang.AssertionError: There should be only 1 data file in partition 'aaa' expected:<1> but was:<2>
        at org.junit.Assert.fail(Assert.java:88)
        at org.junit.Assert.failNotEquals(Assert.java:834)
        at org.junit.Assert.assertEquals(Assert.java:645)
        at org.apache.iceberg.flink.TestFlinkTableSink.testHashDistributeMode(TestFlinkTableSink.java:274)

    org.apache.flink.table.api.ValidationException: Could not execute DROP DATABASE IF EXISTS  testhive.db RESTRICT
        at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeOperation(TableEnvironmentImpl.java:989)
        at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:666)
        at org.apache.iceberg.flink.FlinkTestBase.exec(FlinkTestBase.java:91)
        at org.apache.iceberg.flink.FlinkTestBase.exec(FlinkTestBase.java:95)
        at org.apache.iceberg.flink.FlinkTestBase.sql(FlinkTestBase.java:99)
        at org.apache.iceberg.flink.TestFlinkTableSink.clean(TestFlinkTableSink.java:126)

        Caused by:
        org.apache.flink.table.catalog.exceptions.DatabaseNotEmptyException: Database db in catalog testhive is not empty.
            at org.apache.iceberg.flink.FlinkCatalog.dropDatabase(FlinkCatalog.java:240)
            at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeOperation(TableEnvironmentImpl.java:983)
            ... 5 more

            Caused by:
            org.apache.iceberg.exceptions.NamespaceNotEmptyException: Namespace db is not empty. One or more tables exist.
                at org.apache.iceberg.hive.HiveCatalog.dropNamespace(HiveCatalog.java:307)
                at org.apache.iceberg.flink.FlinkCatalog.dropDatabase(FlinkCatalog.java:231)
                ... 6 more

                Caused by:
                InvalidOperationException(message:Database db is not empty. One or more tables exist.)
                    at org.apache.hadoop.hive.metastore.api.ThriftHiveMetastore$drop_database_result$drop_database_resultStandardScheme.read(ThriftHiveMetastore.java:28714)
                    at org.apache.hadoop.hive.metastore.api.ThriftHiveMetastore$drop_database_result$drop_database_resultStandardScheme.read(ThriftHiveMetastore.java:28691)
                    at org.apache.hadoop.hive.metastore.api.ThriftHiveMetastore$drop_database_result.read(ThriftHiveMetastore.java:28625)
                    at org.apache.thrift.TServiceClient.receiveBase(TServiceClient.java:86)
                    at org.apache.hadoop.hive.metastore.api.ThriftHiveMetastore$Client.recv_drop_database(ThriftHiveMetastore.java:813)
                    at org.apache.hadoop.hive.metastore.api.ThriftHiveMetastore$Client.drop_database(ThriftHiveMetastore.java:798)
                    at org.apache.hadoop.hive.metastore.HiveMetaStoreClient.dropDatabase(HiveMetaStoreClient.java:868)
                    at org.apache.iceberg.hive.HiveCatalog.lambda$dropNamespace$9(HiveCatalog.java:296)
                    at org.apache.iceberg.ClientPoolImpl.run(ClientPoolImpl.java:51)
                    at org.apache.iceberg.hive.CachedClientPool.run(CachedClientPool.java:77)
                    at org.apache.iceberg.hive.HiveCatalog.dropNamespace(HiveCatalog.java:295)
                    ... 7 more

[nastra]

I have been running into this quite frequently in the past, so I disabled it in #3307 until it's fixed

[yittg]

I encountered this failed test case twice, and i think i reproduced it locally.

Before diving it deeply, i think it's better to share the log here,
and at first glance it looks like it is different from the conclusion.

The following is the detail:

[Source: Values(tuples=[[{ 1, _UTF-16LE'aaa' }, { 1, _UTF-16LE'bbb' }, { 1, _UTF-16LE'ccc' }, { 2, _UTF-16LE'aaa' }, { 2, _UTF-16LE'bbb' }, { 2, _UTF-16LE'ccc' }, { 3, _UTF-16LE'aaa' }, { 3, _UTF-16LE'bbb' }, { 3, _UTF-16LE'ccc' }]]) (1/1)#0] INFO org.apache.flink.runtime.taskmanager.Task - Source: Values(tuples=[[{ 1, _UTF-16LE'aaa' }, { 1, _UTF-16LE'bbb' }, { 1, _UTF-16LE'ccc' }, { 2, _UTF-16LE'aaa' }, { 2, _UTF-16LE'bbb' }, { 2, _UTF-16LE'ccc' }, { 3, _UTF-16LE'aaa' }, { 3, _UTF-16LE'bbb' }, { 3, _UTF-16LE'ccc' }]]) (1/1)#0 (c3d03556514594e8aff0175bbd12d35e) switched from INITIALIZING to RUNNING.
[flink-akka.actor.default-dispatcher-8] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Source: Values(tuples=[[{ 1, _UTF-16LE'aaa' }, { 1, _UTF-16LE'bbb' }, { 1, _UTF-16LE'ccc' }, { 2, _UTF-16LE'aaa' }, { 2, _UTF-16LE'bbb' }, { 2, _UTF-16LE'ccc' }, { 3, _UTF-16LE'aaa' }, { 3, _UTF-16LE'bbb' }, { 3, _UTF-16LE'ccc' }]]) (1/1) (c3d03556514594e8aff0175bbd12d35e) switched from INITIALIZING to RUNNING.
[IcebergStreamWriter (1/1)#0] INFO org.apache.flink.runtime.state.heap.HeapKeyedStateBackend - Initializing heap keyed state backend with stream factory.
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.iceberg.BaseMetastoreTableOperations - Refreshing table metadata from new version: file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit9733325594032949849/db.db/test_hash_distribution_mode/metadata/00000-e773c6cb-c67a-422d-9c56-8c68d3d2d64b.metadata.json
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.iceberg.BaseMetastoreCatalog - Table loaded by catalog: testhive.db.test_hash_distribution_mode
[IcebergStreamWriter (1/1)#0] INFO org.apache.flink.runtime.taskmanager.Task - IcebergStreamWriter (1/1)#0 (e863df4f3c93498a6d45488a9898774b) switched from INITIALIZING to RUNNING.
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - IcebergStreamWriter (1/1) (e863df4f3c93498a6d45488a9898774b) switched from INITIALIZING to RUNNING.
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.flink.runtime.taskmanager.Task - IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0 (4154094a26a2ffc60623d0ec10172143) switched from INITIALIZING to RUNNING.
[flink-akka.actor.default-dispatcher-8] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1) (4154094a26a2ffc60623d0ec10172143) switched from INITIALIZING to RUNNING.
[Checkpoint Timer] INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Triggering checkpoint 1 (type=CHECKPOINT) @ 1645431273362 for job 32b28d9a2d686b0cb1ed6efb940781b5.
[Source: Values(tuples=[[{ 1, _UTF-16LE'aaa' }, { 1, _UTF-16LE'bbb' }, { 1, _UTF-16LE'ccc' }, { 2, _UTF-16LE'aaa' }, { 2, _UTF-16LE'bbb' }, { 2, _UTF-16LE'ccc' }, { 3, _UTF-16LE'aaa' }, { 3, _UTF-16LE'bbb' }, { 3, _UTF-16LE'ccc' }]]) (1/1)#0] INFO org.apache.flink.runtime.taskmanager.Task - Source: Values(tuples=[[{ 1, _UTF-16LE'aaa' }, { 1, _UTF-16LE'bbb' }, { 1, _UTF-16LE'ccc' }, { 2, _UTF-16LE'aaa' }, { 2, _UTF-16LE'bbb' }, { 2, _UTF-16LE'ccc' }, { 3, _UTF-16LE'aaa' }, { 3, _UTF-16LE'bbb' }, { 3, _UTF-16LE'ccc' }]]) (1/1)#0 (c3d03556514594e8aff0175bbd12d35e) switched from RUNNING to FINISHED.
[Source: Values(tuples=[[{ 1, _UTF-16LE'aaa' }, { 1, _UTF-16LE'bbb' }, { 1, _UTF-16LE'ccc' }, { 2, _UTF-16LE'aaa' }, { 2, _UTF-16LE'bbb' }, { 2, _UTF-16LE'ccc' }, { 3, _UTF-16LE'aaa' }, { 3, _UTF-16LE'bbb' }, { 3, _UTF-16LE'ccc' }]]) (1/1)#0] INFO org.apache.flink.runtime.taskmanager.Task - Freeing task resources for Source: Values(tuples=[[{ 1, _UTF-16LE'aaa' }, { 1, _UTF-16LE'bbb' }, { 1, _UTF-16LE'ccc' }, { 2, _UTF-16LE'aaa' }, { 2, _UTF-16LE'bbb' }, { 2, _UTF-16LE'ccc' }, { 3, _UTF-16LE'aaa' }, { 3, _UTF-16LE'bbb' }, { 3, _UTF-16LE'ccc' }]]) (1/1)#0 (c3d03556514594e8aff0175bbd12d35e).
[flink-akka.actor.default-dispatcher-8] INFO org.apache.flink.runtime.taskexecutor.TaskExecutor - Un-registering task and sending final execution state FINISHED to JobManager for task Source: Values(tuples=[[{ 1, _UTF-16LE'aaa' }, { 1, _UTF-16LE'bbb' }, { 1, _UTF-16LE'ccc' }, { 2, _UTF-16LE'aaa' }, { 2, _UTF-16LE'bbb' }, { 2, _UTF-16LE'ccc' }, { 3, _UTF-16LE'aaa' }, { 3, _UTF-16LE'bbb' }, { 3, _UTF-16LE'ccc' }]]) (1/1)#0 c3d03556514594e8aff0175bbd12d35e.
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Source: Values(tuples=[[{ 1, _UTF-16LE'aaa' }, { 1, _UTF-16LE'bbb' }, { 1, _UTF-16LE'ccc' }, { 2, _UTF-16LE'aaa' }, { 2, _UTF-16LE'bbb' }, { 2, _UTF-16LE'ccc' }, { 3, _UTF-16LE'aaa' }, { 3, _UTF-16LE'bbb' }, { 3, _UTF-16LE'ccc' }]]) (1/1) (c3d03556514594e8aff0175bbd12d35e) switched from RUNNING to FINISHED.
[IcebergStreamWriter (1/1)#0] INFO org.apache.orc.impl.HadoopShimsPre2_7 - Can't get KeyProvider for ORC encryption from hadoop.security.key.provider.path.
[IcebergStreamWriter (1/1)#0] INFO org.apache.orc.impl.PhysicalFsWriter - ORC writer created for path: file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit9733325594032949849/db.db/test_hash_distribution_mode/data/data=aaa/00000-0-0e650bad-9e4f-4953-b2f6-3a99868aa38a-00001.orc with stripeSize: 67108864 blockSize: 268435456 compression: Compress: ZLIB buffer: 262144
[IcebergStreamWriter (1/1)#0] INFO org.apache.orc.impl.WriterImpl - ORC writer created for path: file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit9733325594032949849/db.db/test_hash_distribution_mode/data/data=aaa/00000-0-0e650bad-9e4f-4953-b2f6-3a99868aa38a-00001.orc with stripeSize: 67108864 options: Compress: ZLIB buffer: 262144
[IcebergStreamWriter (1/1)#0] INFO org.apache.orc.impl.PhysicalFsWriter - ORC writer created for path: file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit9733325594032949849/db.db/test_hash_distribution_mode/data/data=bbb/00000-0-0e650bad-9e4f-4953-b2f6-3a99868aa38a-00002.orc with stripeSize: 67108864 blockSize: 268435456 compression: Compress: ZLIB buffer: 262144
[IcebergStreamWriter (1/1)#0] INFO org.apache.orc.impl.WriterImpl - ORC writer created for path: file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit9733325594032949849/db.db/test_hash_distribution_mode/data/data=bbb/00000-0-0e650bad-9e4f-4953-b2f6-3a99868aa38a-00002.orc with stripeSize: 67108864 options: Compress: ZLIB buffer: 262144
[IcebergStreamWriter (1/1)#0] INFO org.apache.orc.impl.PhysicalFsWriter - ORC writer created for path: file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit9733325594032949849/db.db/test_hash_distribution_mode/data/data=ccc/00000-0-0e650bad-9e4f-4953-b2f6-3a99868aa38a-00003.orc with stripeSize: 67108864 blockSize: 268435456 compression: Compress: ZLIB buffer: 262144
[IcebergStreamWriter (1/1)#0] INFO org.apache.orc.impl.WriterImpl - ORC writer created for path: file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit9733325594032949849/db.db/test_hash_distribution_mode/data/data=ccc/00000-0-0e650bad-9e4f-4953-b2f6-3a99868aa38a-00003.orc with stripeSize: 67108864 options: Compress: ZLIB buffer: 262144
[IcebergStreamWriter (1/1)#0] INFO org.apache.orc.impl.PhysicalFsWriter - ORC writer created for path: file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit9733325594032949849/db.db/test_hash_distribution_mode/data/data=aaa/00000-0-0e650bad-9e4f-4953-b2f6-3a99868aa38a-00004.orc with stripeSize: 67108864 blockSize: 268435456 compression: Compress: ZLIB buffer: 262144
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.iceberg.flink.sink.IcebergFilesCommitter - Start to flush snapshot state to state backend, table: testhive.db.test_hash_distribution_mode, checkpointId: 1
[IcebergStreamWriter (1/1)#0] INFO org.apache.orc.impl.WriterImpl - ORC writer created for path: file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit9733325594032949849/db.db/test_hash_distribution_mode/data/data=aaa/00000-0-0e650bad-9e4f-4953-b2f6-3a99868aa38a-00004.orc with stripeSize: 67108864 options: Compress: ZLIB buffer: 262144
[IcebergStreamWriter (1/1)#0] INFO org.apache.flink.runtime.taskmanager.Task - IcebergStreamWriter (1/1)#0 (e863df4f3c93498a6d45488a9898774b) switched from RUNNING to FINISHED.
[IcebergStreamWriter (1/1)#0] INFO org.apache.flink.runtime.taskmanager.Task - Freeing task resources for IcebergStreamWriter (1/1)#0 (e863df4f3c93498a6d45488a9898774b).
[flink-akka.actor.default-dispatcher-8] INFO org.apache.flink.runtime.taskexecutor.TaskExecutor - Un-registering task and sending final execution state FINISHED to JobManager for task IcebergStreamWriter (1/1)#0 e863df4f3c93498a6d45488a9898774b.
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - IcebergStreamWriter (1/1) (e863df4f3c93498a6d45488a9898774b) switched from RUNNING to FINISHED.
[jobmanager-io-thread-5] INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed checkpoint 1 for job 32b28d9a2d686b0cb1ed6efb940781b5 (3397 bytes, checkpointDuration=908 ms, finalizationTime=4 ms).
[Checkpoint Timer] INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Failed to trigger checkpoint for job 32b28d9a2d686b0cb1ed6efb940781b5 because Some tasks of the job have already finished and checkpointing with finished tasks is not enabled. Failure reason: Not all required tasks are currently running.
[Checkpoint Timer] INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Failed to trigger checkpoint for job 32b28d9a2d686b0cb1ed6efb940781b5 because Some tasks of the job have already finished and checkpointing with finished tasks is not enabled. Failure reason: Not all required tasks are currently running.
[Checkpoint Timer] INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Failed to trigger checkpoint for job 32b28d9a2d686b0cb1ed6efb940781b5 because Some tasks of the job have already finished and checkpointing with finished tasks is not enabled. Failure reason: Not all required tasks are currently running.
[Checkpoint Timer] INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Failed to trigger checkpoint for job 32b28d9a2d686b0cb1ed6efb940781b5 because Some tasks of the job have already finished and checkpointing with finished tasks is not enabled. Failure reason: Not all required tasks are currently running.
[Checkpoint Timer] INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Failed to trigger checkpoint for job 32b28d9a2d686b0cb1ed6efb940781b5 because Some tasks of the job have already finished and checkpointing with finished tasks is not enabled. Failure reason: Not all required tasks are currently running.
[Checkpoint Timer] INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Failed to trigger checkpoint for job 32b28d9a2d686b0cb1ed6efb940781b5 because Some tasks of the job have already finished and checkpointing with finished tasks is not enabled. Failure reason: Not all required tasks are currently running.
[Checkpoint Timer] INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Failed to trigger checkpoint for job 32b28d9a2d686b0cb1ed6efb940781b5 because Some tasks of the job have already finished and checkpointing with finished tasks is not enabled. Failure reason: Not all required tasks are currently running.
[Checkpoint Timer] INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Failed to trigger checkpoint for job 32b28d9a2d686b0cb1ed6efb940781b5 because Some tasks of the job have already finished and checkpointing with finished tasks is not enabled. Failure reason: Not all required tasks are currently running.
[Checkpoint Timer] INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Failed to trigger checkpoint for job 32b28d9a2d686b0cb1ed6efb940781b5 because Some tasks of the job have already finished and checkpointing with finished tasks is not enabled. Failure reason: Not all required tasks are currently running.
[Checkpoint Timer] INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Failed to trigger checkpoint for job 32b28d9a2d686b0cb1ed6efb940781b5 because Some tasks of the job have already finished and checkpointing with finished tasks is not enabled. Failure reason: Not all required tasks are currently running.
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.iceberg.flink.sink.IcebergFilesCommitter - Committing append with 4 data files and 0 delete files to table testhive.db.test_hash_distribution_mode
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode	
[Checkpoint Timer] INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Failed to trigger checkpoint for job 32b28d9a2d686b0cb1ed6efb940781b5 because Some tasks of the job have already finished and checkpointing with finished tasks is not enabled. Failure reason: Not all required tasks are currently running.
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode	
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 alter_table: db=db tbl=test_hash_distribution_mode newtbl=test_hash_distribution_mode
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 alter_table: db=db tbl=test_hash_distribution_mode newtbl=test_hash_distribution_mode	
[Checkpoint Timer] INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Failed to trigger checkpoint for job 32b28d9a2d686b0cb1ed6efb940781b5 because Some tasks of the job have already finished and checkpointing with finished tasks is not enabled. Failure reason: Not all required tasks are currently running.
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.iceberg.BaseMetastoreTableOperations - Successfully committed to table testhive.db.test_hash_distribution_mode in 105 ms
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.iceberg.SnapshotProducer - Committed snapshot 1987697929173874507 (MergeAppend)
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode	
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.iceberg.BaseMetastoreTableOperations - Refreshing table metadata from new version: file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit9733325594032949849/db.db/test_hash_distribution_mode/metadata/00001-6d7501d3-28cb-4ec2-bdb8-51c3b948589e.metadata.json
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode	
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.iceberg.flink.sink.IcebergFilesCommitter - Committed in 225 ms
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.flink.runtime.taskmanager.Task - IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0 (4154094a26a2ffc60623d0ec10172143) switched from RUNNING to FINISHED.
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.flink.runtime.taskmanager.Task - Freeing task resources for IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0 (4154094a26a2ffc60623d0ec10172143).
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.taskexecutor.TaskExecutor - Un-registering task and sending final execution state FINISHED to JobManager for task IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0 4154094a26a2ffc60623d0ec10172143.
[flink-akka.actor.default-dispatcher-8] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1) (4154094a26a2ffc60623d0ec10172143) switched from RUNNING to FINISHED.
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.resourcemanager.slotmanager.DeclarativeSlotManager - Clearing resource requirements of job 32b28d9a2d686b0cb1ed6efb940781b5
[flink-akka.actor.default-dispatcher-8] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Job insert-into_testhive.db.test_hash_distribution_mode (32b28d9a2d686b0cb1ed6efb940781b5) switched from state RUNNING to FINISHED.
[flink-akka.actor.default-dispatcher-8] INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Stopping checkpoint coordinator for job 32b28d9a2d686b0cb1ed6efb940781b5.
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.dispatcher.StandaloneDispatcher - Job 32b28d9a2d686b0cb1ed6efb940781b5 reached terminal state FINISHED.
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode	
[flink-akka.actor.default-dispatcher-8] INFO org.apache.flink.runtime.jobmaster.JobMaster - Stopping the JobMaster for job 'insert-into_testhive.db.test_hash_distribution_mode' (32b28d9a2d686b0cb1ed6efb940781b5).
[flink-akka.actor.default-dispatcher-8] INFO org.apache.flink.runtime.checkpoint.StandaloneCompletedCheckpointStore - Shutting down

The result as expected

There should be 1 data file in partition 'aaa' expected:<1> but was:<2>
Expected :1
Actual   :2
<Click to see difference>

java.lang.AssertionError: There should be 1 data file in partition 'aaa' expected:<1> but was:<2>

[openinx]

Thanks for the detailed log message, Did you find the way to reproduce this failure in you local host? ( I still can not reproduce).

I read the log message carefully, it's quite strange that producing 4 data files in a single checkpoint.

➜  ~ cat Untitled-2 | grep 'org.apache.orc.impl.PhysicalFsWriter - ORC writer created for path'
[IcebergStreamWriter (1/1)#0] INFO org.apache.orc.impl.PhysicalFsWriter - ORC writer created for path: file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit9733325594032949849/db.db/test_hash_distribution_mode/data/data=aaa/00000-0-0e650bad-9e4f-4953-b2f6-3a99868aa38a-00001.orc with stripeSize: 67108864 blockSize: 268435456 compression: Compress: ZLIB buffer: 262144
[IcebergStreamWriter (1/1)#0] INFO org.apache.orc.impl.PhysicalFsWriter - ORC writer created for path: file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit9733325594032949849/db.db/test_hash_distribution_mode/data/data=bbb/00000-0-0e650bad-9e4f-4953-b2f6-3a99868aa38a-00002.orc with stripeSize: 67108864 blockSize: 268435456 compression: Compress: ZLIB buffer: 262144
[IcebergStreamWriter (1/1)#0] INFO org.apache.orc.impl.PhysicalFsWriter - ORC writer created for path: file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit9733325594032949849/db.db/test_hash_distribution_mode/data/data=ccc/00000-0-0e650bad-9e4f-4953-b2f6-3a99868aa38a-00003.orc with stripeSize: 67108864 blockSize: 268435456 compression: Compress: ZLIB buffer: 262144
[IcebergStreamWriter (1/1)#0] INFO org.apache.orc.impl.PhysicalFsWriter - ORC writer created for path: file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit9733325594032949849/db.db/test_hash_distribution_mode/data/data=aaa/00000-0-0e650bad-9e4f-4953-b2f6-3a99868aa38a-00004.orc with stripeSize: 67108864 blockSize: 268435456 compression: Compress: ZLIB buffer: 262144

Only the partition 'aaa' produces two orc data files.

[yittg]

@openinx i can not reproduce it stably for now,
I think you can just repeat it multiple times, for example, by generating thousands of parameters, or script.
And i think decreasing the checkpoint interval can help reproducing it, for example, to 100ms.

[openinx]

Here we have only 1 parallelism in this unit test to write records in this stream flow:

Source -> Shuffle -> IcebergStreamWriter -> IcebergFilesCommitter.

And we have the following records that need to write into the partitioned table (partition field is data):

(1, 'aaa'), (1, 'bbb'), (1, 'ccc')
(2, 'aaa'), (2, 'bbb'), (2, 'ccc')
(3, 'aaa'), (3, 'bbb'), (3, 'ccc')

Then:

Step#1 we write the following records into orc files:

(1, 'aaa'), (1, 'bbb'), (1, 'ccc')
(2, 'aaa'), (2, 'bbb'), (2, 'ccc')
(3, 'bbb'), (3, 'ccc')

# Notice: the record (3, 'aaa') was not emitted to the orc file in the checkpoint#1 

Step#2 checkpoint barrier was encountered, so the IcebergStreamWriter emits the 1.orc, 2.orc, 3.orc to IcebergFilesCommitter.
Step#3 IcebergFilesCommitter was trying to execute snapshotState;
Step#4 The IcebergStreamWriter emitted a 4.orc with (3, 'aaa') included (endInput), and then close itself.
Step#5 The IcebergFilesCommitter commit the transaction with 4 orc data files. (notifyCheckpointComplete)

The log message seems did the above steps, and finally commit all the 4 orc data files in a single transaction. But in fact, the flink IcebergFilesCommitter won't accept any new file (4.orc) to the pending transaction when it is executing the snapshotState in Step#3 because the flink's StreamTask is a single thread consuming message from a FIFO queue. I think there must be other wrong thing but I cannot reproduce this thing.

Anyway, the root cause for this failure unit test is: we don't control all the events precisely in a single checkpoint (In this case, few records are accumulated in one checkpoint, but there is still someone which was remained for the endInput to emit). So I think the solution I proposed in this #4117 (comment) should still work to fix it fundamentally.

[yittg]

The following test can easily failed, i think it is equivalent to the original IIUC. It will lead to different kinds of error:

1. Committing with more than one files in one snapshot;
2. Committing with one file in each snapshot, but failed on the final assert.

Hope it can help, @openinx.

  @Test
  public void testHashDistributeMode() throws Exception {
    String tableName = "test_hash_distribution_mode";
    Map<String, String> tableProps = ImmutableMap.of(
        "write.format.default", format.name(),
        TableProperties.WRITE_DISTRIBUTION_MODE, DistributionMode.HASH.modeName()
    );
    sql("CREATE TABLE default_catalog.default_database.src (id INT) WITH %s",
        toWithClause(ImmutableMap.of(
            "connector","datagen",
            "number-of-rows", "100000",
            "rows-per-second", "100000",
            "fields.id.kind", "sequence",
            "fields.id.start", "1",
            "fields.id.end", "100000")));
    sql("CREATE TABLE %s(id INT, data VARCHAR) PARTITIONED BY (data) WITH %s",
        tableName, toWithClause(tableProps));

    try {
      // Insert data set.

      sql("INSERT INTO %s SELECT id, 'aaa' as data FROM default_catalog.default_database.src", tableName);

      Table table = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, tableName));

      // Sometimes we will have more than one checkpoint if we pass the auto checkpoint interval,
      // thus producing multiple snapshots.  Here we assert that each snapshot has only 1 file per partition.
      Map<Long, List<DataFile>> snapshotToDataFiles = SimpleDataUtil.snapshotToDataFiles(table);
      for (List<DataFile> dataFiles : snapshotToDataFiles.values()) {
        Assert.assertEquals("There should be 1 data file in partition 'aaa'", 1,
            SimpleDataUtil.matchingPartitions(dataFiles, table.spec(), ImmutableMap.of("data", "aaa")).size());
      }
    } finally {
      sql("DROP TABLE IF EXISTS %s.%s", flinkDatabase, tableName);
    }
  }

[yittg]

2. Committing with one file in each snapshot, but failed on the final assert.

This can be fixed by the following change:

--- a/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java
+++ b/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/SimpleDataUtil.java
@@ -275,10 +275,10 @@ public class SimpleDataUtil {
       TableScan tableScan = table.newScan();
       if (current.parentId() != null) {
         // Collect the data files that was added only in current snapshot.
-        tableScan.appendsBetween(current.parentId(), current.snapshotId());
+        tableScan = tableScan.appendsBetween(current.parentId(), current.snapshotId());
       } else {
         // Collect the data files that was added in the oldest snapshot.
-        tableScan.useSnapshot(current.snapshotId());
+        tableScan = tableScan.useSnapshot(current.snapshotId());
       }

[yittg]

1. Committing with more than one files in one snapshot;

For part-1, i think the following log is more detailed by adding some flag log,
It's mostly like because EndInput comes before notifyCheckpointComplete. With the following timeline:
W is for IcebergStreamWriter, C is for IcebergFilesCommitter

• [W]file-1 created;
• [W]checkpoint-1 prepared;
• [W]result with file-1 emitted;
• [W]file-2 created;
• [C]checkpoint-1 notified, snapshot-1 with file-1 committed;
• [W]checkpoint-2 prepared;
• [W]result with file-2 emitted;
• [W] file-3 created;
• [W] endInput;
• [W] result with file-3 emitted;
• [C] endInput, snapshot-2 with file-2, file-3 committed;
• [C] checkpoint-2 notified;

--- a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java
+++ b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java
  public void notifyCheckpointComplete(long checkpointId) throws Exception {
+    LOG.info("Checkpoint notified, #{}", checkpointId);
     if (checkpointId > maxCommittedCheckpointId) {
       commitUpToCheckpoint(dataFilesPerCheckpoint, flinkJobId, checkpointId);
       this.maxCommittedCheckpointId = checkpointId;
...
   public void endInput() throws IOException {
+    LOG.info("End input reached");
     // Flush the buffered data files into 'dataFilesPerCheckpoint' firstly.
     long currentCheckpointId = Long.MAX_VALUE;
     dataFilesPerCheckpoint.put(currentCheckpointId, writeToManifest(currentCheckpointId));
--- a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriter.java
+++ b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriter.java
   @Override
   public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
+    LOG.info("Before checkpoint #{}", checkpointId);
     // close all open files and emit files to downstream committer operator
     emit(writer.complete());
...
   @Override
   public void endInput() throws IOException {
+    LOG.info("End input reached");
     // For bounded stream, it may don't enable the checkpoint mechanism so we'd better to emit the remaining
     // completed files to downstream before closing the writer so that we won't miss any of them.
     emit(writer.complete());

[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_table : db=db tbl=test_table
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_table : db=db tbl=test_table	
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 create_table: Table(tableName:test_table, dbName:db, owner:tangyi, createTime:540425, lastAccessTime:540425, retention:2147483647, sd:StorageDescriptor(cols:[FieldSchema(name:id, type:int, comment:null), FieldSchema(name:data, type:string, comment:null)], location:file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit13141381595252095998/db.db/test_table, inputFormat:org.apache.hadoop.mapred.FileInputFormat, outputFormat:org.apache.hadoop.mapred.FileOutputFormat, compressed:false, numBuckets:0, serdeInfo:SerDeInfo(name:null, serializationLib:org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, parameters:{}), bucketCols:null, sortCols:null, parameters:null), partitionKeys:[], parameters:{EXTERNAL=TRUE, write.format.default=ORC, metadata_location=file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit13141381595252095998/db.db/test_table/metadata/00000-b94688ab-1fd9-465a-bed0-e4b1a93c98fb.metadata.json, uuid=ebd3ca64-66b1-40b1-8a6c-6b356dcd1e39, table_type=ICEBERG}, viewOriginalText:null, viewExpandedText:null, tableType:EXTERNAL_TABLE)
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 create_table: Table(tableName:test_table, dbName:db, owner:tangyi, createTime:540425, lastAccessTime:540425, retention:2147483647, sd:StorageDescriptor(cols:[FieldSchema(name:id, type:int, comment:null), FieldSchema(name:data, type:string, comment:null)], location:file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit13141381595252095998/db.db/test_table, inputFormat:org.apache.hadoop.mapred.FileInputFormat, outputFormat:org.apache.hadoop.mapred.FileOutputFormat, compressed:false, numBuckets:0, serdeInfo:SerDeInfo(name:null, serializationLib:org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, parameters:{}), bucketCols:null, sortCols:null, parameters:null), partitionKeys:[], parameters:{EXTERNAL=TRUE, write.format.default=ORC, metadata_location=file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit13141381595252095998/db.db/test_table/metadata/00000-b94688ab-1fd9-465a-bed0-e4b1a93c98fb.metadata.json, uuid=ebd3ca64-66b1-40b1-8a6c-6b356dcd1e39, table_type=ICEBERG}, viewOriginalText:null, viewExpandedText:null, tableType:EXTERNAL_TABLE)	
[pool-11-thread-1] INFO hive.log - Updating table stats fast for test_table
[pool-11-thread-1] INFO hive.log - Updated size of table test_table to 1245
[Test worker] INFO org.apache.iceberg.BaseMetastoreTableOperations - Successfully committed to table testhive.db.test_table in 809 ms
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_table : db=db tbl=test_table
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_table : db=db tbl=test_table	
[Test worker] INFO org.apache.iceberg.BaseMetastoreTableOperations - Refreshing table metadata from new version: file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit13141381595252095998/db.db/test_table/metadata/00000-b94688ab-1fd9-465a-bed0-e4b1a93c98fb.metadata.json
[Test worker] INFO org.apache.iceberg.BaseMetastoreCatalog - Table loaded by catalog: hive.db.test_table
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode	
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_database: db
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_database: db	
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode	
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 create_table: Table(tableName:test_hash_distribution_mode, dbName:db, owner:tangyi, createTime:540425, lastAccessTime:540425, retention:2147483647, sd:StorageDescriptor(cols:[FieldSchema(name:id, type:int, comment:null), FieldSchema(name:data, type:string, comment:null)], location:file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit13141381595252095998/db.db/test_hash_distribution_mode, inputFormat:org.apache.hadoop.mapred.FileInputFormat, outputFormat:org.apache.hadoop.mapred.FileOutputFormat, compressed:false, numBuckets:0, serdeInfo:SerDeInfo(name:null, serializationLib:org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, parameters:{}), bucketCols:null, sortCols:null, parameters:null), partitionKeys:[], parameters:{EXTERNAL=TRUE, write.format.default=ORC, metadata_location=file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit13141381595252095998/db.db/test_hash_distribution_mode/metadata/00000-12b9361e-35a0-4862-94ec-9bae7deb0fe1.metadata.json, write.distribution-mode=hash, uuid=f2be590f-787b-4bf9-a4fa-4e8339d4e1a6, table_type=ICEBERG}, viewOriginalText:null, viewExpandedText:null, tableType:EXTERNAL_TABLE)
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 create_table: Table(tableName:test_hash_distribution_mode, dbName:db, owner:tangyi, createTime:540425, lastAccessTime:540425, retention:2147483647, sd:StorageDescriptor(cols:[FieldSchema(name:id, type:int, comment:null), FieldSchema(name:data, type:string, comment:null)], location:file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit13141381595252095998/db.db/test_hash_distribution_mode, inputFormat:org.apache.hadoop.mapred.FileInputFormat, outputFormat:org.apache.hadoop.mapred.FileOutputFormat, compressed:false, numBuckets:0, serdeInfo:SerDeInfo(name:null, serializationLib:org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, parameters:{}), bucketCols:null, sortCols:null, parameters:null), partitionKeys:[], parameters:{EXTERNAL=TRUE, write.format.default=ORC, metadata_location=file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit13141381595252095998/db.db/test_hash_distribution_mode/metadata/00000-12b9361e-35a0-4862-94ec-9bae7deb0fe1.metadata.json, write.distribution-mode=hash, uuid=f2be590f-787b-4bf9-a4fa-4e8339d4e1a6, table_type=ICEBERG}, viewOriginalText:null, viewExpandedText:null, tableType:EXTERNAL_TABLE)	
[pool-11-thread-1] INFO hive.log - Updating table stats fast for test_hash_distribution_mode
[pool-11-thread-1] INFO hive.log - Updated size of table test_hash_distribution_mode to 1513
[Test worker] INFO org.apache.iceberg.BaseMetastoreTableOperations - Successfully committed to table testhive.db.test_hash_distribution_mode in 93 ms
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_database: db
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_database: db	
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_table : db=db tbl=default_catalog
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_table : db=db tbl=default_catalog	
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_database: default_catalog
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_database: default_catalog	
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_database: db
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_database: db	
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_database: default_catalog
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_database: default_catalog	
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode	
[Test worker] INFO org.apache.iceberg.BaseMetastoreTableOperations - Refreshing table metadata from new version: file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit13141381595252095998/db.db/test_hash_distribution_mode/metadata/00000-12b9361e-35a0-4862-94ec-9bae7deb0fe1.metadata.json
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode	
[Test worker] INFO org.apache.iceberg.BaseMetastoreTableOperations - Refreshing table metadata from new version: file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit13141381595252095998/db.db/test_hash_distribution_mode/metadata/00000-12b9361e-35a0-4862-94ec-9bae7deb0fe1.metadata.json
[Test worker] INFO org.apache.iceberg.BaseMetastoreCatalog - Table loaded by catalog: testhive.db.test_hash_distribution_mode
[Test worker] INFO org.apache.flink.api.java.typeutils.TypeExtractor - class org.apache.iceberg.io.WriteResult does not contain a setter for field dataFiles
[Test worker] INFO org.apache.flink.api.java.typeutils.TypeExtractor - Class class org.apache.iceberg.io.WriteResult cannot be used as a POJO type because not all fields are valid POJO fields, and must be processed as GenericType. Please read the Flink documentation on "Data Types & Serialization" for details of the effect on performance.
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.dispatcher.StandaloneDispatcher - Received JobGraph submission 'insert-into_testhive.db.test_hash_distribution_mode' (0462ddc2e6a8276f79b3914f9b4fcd19).
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.dispatcher.StandaloneDispatcher - Submitting job 'insert-into_testhive.db.test_hash_distribution_mode' (0462ddc2e6a8276f79b3914f9b4fcd19).
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.highavailability.nonha.embedded.EmbeddedLeaderService - Proposing leadership to contender LeaderContender: JobMasterServiceLeadershipRunner
[jobmanager-io-thread-1] INFO org.apache.flink.runtime.rpc.akka.AkkaRpcService - Starting RPC endpoint for org.apache.flink.runtime.jobmaster.JobMaster at akka://flink/user/rpc/jobmanager_6 .
[jobmanager-io-thread-1] INFO org.apache.flink.runtime.jobmaster.JobMaster - Initializing job 'insert-into_testhive.db.test_hash_distribution_mode' (0462ddc2e6a8276f79b3914f9b4fcd19).
[jobmanager-io-thread-1] INFO org.apache.flink.runtime.jobmaster.JobMaster - Using restart back off time strategy FixedDelayRestartBackoffTimeStrategy(maxNumberRestartAttempts=2147483647, backoffTimeMS=1000) for insert-into_testhive.db.test_hash_distribution_mode (0462ddc2e6a8276f79b3914f9b4fcd19).
[jobmanager-io-thread-1] INFO org.apache.flink.runtime.jobmaster.JobMaster - Running initialization on master for job insert-into_testhive.db.test_hash_distribution_mode (0462ddc2e6a8276f79b3914f9b4fcd19).
[jobmanager-io-thread-1] INFO org.apache.flink.runtime.jobmaster.JobMaster - Successfully ran initialization on master in 0 ms.
[jobmanager-io-thread-1] INFO org.apache.flink.runtime.scheduler.adapter.DefaultExecutionTopology - Built 1 pipelined regions in 0 ms
[jobmanager-io-thread-1] INFO org.apache.flink.runtime.jobmaster.JobMaster - No state backend has been configured, using default (HashMap) org.apache.flink.runtime.state.hashmap.HashMapStateBackend@2df480e8
[jobmanager-io-thread-1] INFO org.apache.flink.runtime.state.StateBackendLoader - State backend loader loads the state backend as HashMapStateBackend
[jobmanager-io-thread-1] INFO org.apache.flink.runtime.jobmaster.JobMaster - Checkpoint storage is set to 'jobmanager'
[jobmanager-io-thread-1] INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - No checkpoint found during restore.
[jobmanager-io-thread-1] INFO org.apache.flink.runtime.jobmaster.JobMaster - Using failover strategy org.apache.flink.runtime.executiongraph.failover.flip1.RestartPipelinedRegionFailoverStrategy@6a8e2389 for insert-into_testhive.db.test_hash_distribution_mode (0462ddc2e6a8276f79b3914f9b4fcd19).
[jobmanager-io-thread-1] INFO org.apache.flink.runtime.highavailability.nonha.embedded.EmbeddedLeaderService - Received confirmation of leadership for leader akka://flink/user/rpc/jobmanager_6 , session=2cc8799f-155e-4f8f-a8b4-1b9177294dfe
[flink-akka.actor.default-dispatcher-8] INFO org.apache.flink.runtime.jobmaster.JobMaster - Starting execution of job 'insert-into_testhive.db.test_hash_distribution_mode' (0462ddc2e6a8276f79b3914f9b4fcd19) under job master id a8b41b9177294dfe2cc8799f155e4f8f.
[flink-akka.actor.default-dispatcher-8] INFO org.apache.flink.runtime.jobmaster.JobMaster - Starting scheduling with scheduling strategy [org.apache.flink.runtime.scheduler.strategy.PipelinedRegionSchedulingStrategy]
[flink-akka.actor.default-dispatcher-8] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Job insert-into_testhive.db.test_hash_distribution_mode (0462ddc2e6a8276f79b3914f9b4fcd19) switched from state CREATED to RUNNING.
[flink-akka.actor.default-dispatcher-8] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Source: TableSourceScan(table=[[default_catalog, default_database, src]], fields=[id]) -> Calc(select=[id, _UTF-16LE'aaa' AS data]) (1/1) (876164887f156d6a29a5e26a541b1d8b) switched from CREATED to SCHEDULED.
[flink-akka.actor.default-dispatcher-8] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - IcebergStreamWriter (1/1) (c5292a7cfc1d527e13a5eeb9926a4a95) switched from CREATED to SCHEDULED.
[flink-akka.actor.default-dispatcher-8] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1) (ca05ac4af8adb64ada4a789b93be3848) switched from CREATED to SCHEDULED.
[flink-akka.actor.default-dispatcher-8] INFO org.apache.flink.runtime.jobmaster.JobMaster - Connecting to ResourceManager akka://flink/user/rpc/resourcemanager_4(a3ea462904f2b99a88ed07a16e394a8b)
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.jobmaster.JobMaster - Resolved ResourceManager address, beginning registration
[flink-akka.actor.default-dispatcher-5] INFO org.apache.flink.runtime.resourcemanager.StandaloneResourceManager - Registering job manager a8b41b9177294dfe2cc8799f155e4f8f@akka://flink/user/rpc/jobmanager_6 for job 0462ddc2e6a8276f79b3914f9b4fcd19.
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.resourcemanager.StandaloneResourceManager - Registered job manager a8b41b9177294dfe2cc8799f155e4f8f@akka://flink/user/rpc/jobmanager_6 for job 0462ddc2e6a8276f79b3914f9b4fcd19.
[flink-akka.actor.default-dispatcher-5] INFO org.apache.flink.runtime.jobmaster.JobMaster - JobManager successfully registered at ResourceManager, leader id: a3ea462904f2b99a88ed07a16e394a8b.
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.resourcemanager.slotmanager.DeclarativeSlotManager - Received resource requirements from job 0462ddc2e6a8276f79b3914f9b4fcd19: [ResourceRequirement{resourceProfile=ResourceProfile{UNKNOWN}, numberOfRequiredSlots=1}]
[flink-akka.actor.default-dispatcher-5] INFO org.apache.flink.runtime.taskexecutor.TaskExecutor - Receive slot request 55cac133873b326447786e98d4a8bf2b for job 0462ddc2e6a8276f79b3914f9b4fcd19 from resource manager with leader id a3ea462904f2b99a88ed07a16e394a8b.
[flink-akka.actor.default-dispatcher-5] INFO org.apache.flink.runtime.taskexecutor.TaskExecutor - Allocated slot for 55cac133873b326447786e98d4a8bf2b.
[flink-akka.actor.default-dispatcher-5] INFO org.apache.flink.runtime.taskexecutor.DefaultJobLeaderService - Add job 0462ddc2e6a8276f79b3914f9b4fcd19 for job leader monitoring.
[mini-cluster-io-thread-2] INFO org.apache.flink.runtime.taskexecutor.DefaultJobLeaderService - Try to register at job manager akka://flink/user/rpc/jobmanager_6 with leader id 2cc8799f-155e-4f8f-a8b4-1b9177294dfe.
[flink-akka.actor.default-dispatcher-9] INFO org.apache.flink.runtime.taskexecutor.DefaultJobLeaderService - Resolved JobManager address, beginning registration
[flink-akka.actor.default-dispatcher-8] INFO org.apache.flink.runtime.taskexecutor.DefaultJobLeaderService - Successful registration at job manager akka://flink/user/rpc/jobmanager_6 for job 0462ddc2e6a8276f79b3914f9b4fcd19.
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.taskexecutor.TaskExecutor - Establish JobManager connection for job 0462ddc2e6a8276f79b3914f9b4fcd19.
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.taskexecutor.TaskExecutor - Offer reserved slots to the leader of job 0462ddc2e6a8276f79b3914f9b4fcd19.
[flink-akka.actor.default-dispatcher-8] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Source: TableSourceScan(table=[[default_catalog, default_database, src]], fields=[id]) -> Calc(select=[id, _UTF-16LE'aaa' AS data]) (1/1) (876164887f156d6a29a5e26a541b1d8b) switched from SCHEDULED to DEPLOYING.
[flink-akka.actor.default-dispatcher-8] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Deploying Source: TableSourceScan(table=[[default_catalog, default_database, src]], fields=[id]) -> Calc(select=[id, _UTF-16LE'aaa' AS data]) (1/1) (attempt #0) with attempt id 876164887f156d6a29a5e26a541b1d8b to 1948ad9a-cb40-4a3d-9770-db7de5a7bae5 @ localhost (dataPort=-1) with allocation id 55cac133873b326447786e98d4a8bf2b
[flink-akka.actor.default-dispatcher-8] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - IcebergStreamWriter (1/1) (c5292a7cfc1d527e13a5eeb9926a4a95) switched from SCHEDULED to DEPLOYING.
[flink-akka.actor.default-dispatcher-8] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Deploying IcebergStreamWriter (1/1) (attempt #0) with attempt id c5292a7cfc1d527e13a5eeb9926a4a95 to 1948ad9a-cb40-4a3d-9770-db7de5a7bae5 @ localhost (dataPort=-1) with allocation id 55cac133873b326447786e98d4a8bf2b
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.taskexecutor.slot.TaskSlotTableImpl - Activate slot 55cac133873b326447786e98d4a8bf2b.
[flink-akka.actor.default-dispatcher-8] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1) (ca05ac4af8adb64ada4a789b93be3848) switched from SCHEDULED to DEPLOYING.
[flink-akka.actor.default-dispatcher-8] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Deploying IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1) (attempt #0) with attempt id ca05ac4af8adb64ada4a789b93be3848 to 1948ad9a-cb40-4a3d-9770-db7de5a7bae5 @ localhost (dataPort=-1) with allocation id 55cac133873b326447786e98d4a8bf2b
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.state.changelog.StateChangelogStorageLoader - StateChangelogStorageLoader initialized with shortcut names {memory}.
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.state.changelog.StateChangelogStorageLoader - Creating a changelog storage with name 'memory'.
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.taskexecutor.TaskExecutor - Received task Source: TableSourceScan(table=[[default_catalog, default_database, src]], fields=[id]) -> Calc(select=[id, _UTF-16LE'aaa' AS data]) (1/1)#0 (876164887f156d6a29a5e26a541b1d8b), deploy into slot with allocation id 55cac133873b326447786e98d4a8bf2b.
[Source: TableSourceScan(table=[[default_catalog, default_database, src]], fields=[id]) -> Calc(select=[id, _UTF-16LE'aaa' AS data]) (1/1)#0] INFO org.apache.flink.runtime.taskmanager.Task - Source: TableSourceScan(table=[[default_catalog, default_database, src]], fields=[id]) -> Calc(select=[id, _UTF-16LE'aaa' AS data]) (1/1)#0 (876164887f156d6a29a5e26a541b1d8b) switched from CREATED to DEPLOYING.
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.taskexecutor.slot.TaskSlotTableImpl - Activate slot 55cac133873b326447786e98d4a8bf2b.
[Source: TableSourceScan(table=[[default_catalog, default_database, src]], fields=[id]) -> Calc(select=[id, _UTF-16LE'aaa' AS data]) (1/1)#0] INFO org.apache.flink.runtime.taskmanager.Task - Loading JAR files for task Source: TableSourceScan(table=[[default_catalog, default_database, src]], fields=[id]) -> Calc(select=[id, _UTF-16LE'aaa' AS data]) (1/1)#0 (876164887f156d6a29a5e26a541b1d8b) [DEPLOYING].
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.taskexecutor.TaskExecutor - Received task IcebergStreamWriter (1/1)#0 (c5292a7cfc1d527e13a5eeb9926a4a95), deploy into slot with allocation id 55cac133873b326447786e98d4a8bf2b.
[IcebergStreamWriter (1/1)#0] INFO org.apache.flink.runtime.taskmanager.Task - IcebergStreamWriter (1/1)#0 (c5292a7cfc1d527e13a5eeb9926a4a95) switched from CREATED to DEPLOYING.
[IcebergStreamWriter (1/1)#0] INFO org.apache.flink.runtime.taskmanager.Task - Loading JAR files for task IcebergStreamWriter (1/1)#0 (c5292a7cfc1d527e13a5eeb9926a4a95) [DEPLOYING].
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.taskexecutor.slot.TaskSlotTableImpl - Activate slot 55cac133873b326447786e98d4a8bf2b.
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.taskexecutor.TaskExecutor - Received task IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0 (ca05ac4af8adb64ada4a789b93be3848), deploy into slot with allocation id 55cac133873b326447786e98d4a8bf2b.
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.flink.runtime.taskmanager.Task - IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0 (ca05ac4af8adb64ada4a789b93be3848) switched from CREATED to DEPLOYING.
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.flink.runtime.taskmanager.Task - Loading JAR files for task IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0 (ca05ac4af8adb64ada4a789b93be3848) [DEPLOYING].
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.taskexecutor.slot.TaskSlotTableImpl - Activate slot 55cac133873b326447786e98d4a8bf2b.
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.flink.streaming.runtime.tasks.StreamTask - No state backend has been configured, using default (HashMap) org.apache.flink.runtime.state.hashmap.HashMapStateBackend@eb5eb98
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.flink.runtime.state.StateBackendLoader - State backend loader loads the state backend as HashMapStateBackend
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.flink.streaming.runtime.tasks.StreamTask - Checkpoint storage is set to 'jobmanager'
[IcebergStreamWriter (1/1)#0] INFO org.apache.flink.streaming.runtime.tasks.StreamTask - No state backend has been configured, using default (HashMap) org.apache.flink.runtime.state.hashmap.HashMapStateBackend@6873aa5c
[IcebergStreamWriter (1/1)#0] INFO org.apache.flink.runtime.state.StateBackendLoader - State backend loader loads the state backend as HashMapStateBackend
[IcebergStreamWriter (1/1)#0] INFO org.apache.flink.streaming.runtime.tasks.StreamTask - Checkpoint storage is set to 'jobmanager'
[Source: TableSourceScan(table=[[default_catalog, default_database, src]], fields=[id]) -> Calc(select=[id, _UTF-16LE'aaa' AS data]) (1/1)#0] INFO org.apache.flink.streaming.runtime.tasks.StreamTask - No state backend has been configured, using default (HashMap) org.apache.flink.runtime.state.hashmap.HashMapStateBackend@6e0bd0bd
[Source: TableSourceScan(table=[[default_catalog, default_database, src]], fields=[id]) -> Calc(select=[id, _UTF-16LE'aaa' AS data]) (1/1)#0] INFO org.apache.flink.runtime.state.StateBackendLoader - State backend loader loads the state backend as HashMapStateBackend
[Source: TableSourceScan(table=[[default_catalog, default_database, src]], fields=[id]) -> Calc(select=[id, _UTF-16LE'aaa' AS data]) (1/1)#0] INFO org.apache.flink.streaming.runtime.tasks.StreamTask - Checkpoint storage is set to 'jobmanager'
[Checkpoint Timer] INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Failed to trigger checkpoint for job 0462ddc2e6a8276f79b3914f9b4fcd19 because Checkpoint triggering task Source: TableSourceScan(table=[[default_catalog, default_database, src]], fields=[id]) -> Calc(select=[id, _UTF-16LE'aaa' AS data]) (1/1) of job 0462ddc2e6a8276f79b3914f9b4fcd19 is not being executed at the moment. Aborting checkpoint. Failure reason: Not all required tasks are currently running.
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.flink.runtime.taskmanager.Task - IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0 (ca05ac4af8adb64ada4a789b93be3848) switched from DEPLOYING to INITIALIZING.
[Source: TableSourceScan(table=[[default_catalog, default_database, src]], fields=[id]) -> Calc(select=[id, _UTF-16LE'aaa' AS data]) (1/1)#0] INFO org.apache.flink.runtime.taskmanager.Task - Source: TableSourceScan(table=[[default_catalog, default_database, src]], fields=[id]) -> Calc(select=[id, _UTF-16LE'aaa' AS data]) (1/1)#0 (876164887f156d6a29a5e26a541b1d8b) switched from DEPLOYING to INITIALIZING.
[IcebergStreamWriter (1/1)#0] INFO org.apache.flink.runtime.taskmanager.Task - IcebergStreamWriter (1/1)#0 (c5292a7cfc1d527e13a5eeb9926a4a95) switched from DEPLOYING to INITIALIZING.
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1) (ca05ac4af8adb64ada4a789b93be3848) switched from DEPLOYING to INITIALIZING.
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Source: TableSourceScan(table=[[default_catalog, default_database, src]], fields=[id]) -> Calc(select=[id, _UTF-16LE'aaa' AS data]) (1/1) (876164887f156d6a29a5e26a541b1d8b) switched from DEPLOYING to INITIALIZING.
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - IcebergStreamWriter (1/1) (c5292a7cfc1d527e13a5eeb9926a4a95) switched from DEPLOYING to INITIALIZING.
[Source: TableSourceScan(table=[[default_catalog, default_database, src]], fields=[id]) -> Calc(select=[id, _UTF-16LE'aaa' AS data]) (1/1)#0] WARN org.apache.flink.metrics.MetricGroup - The operator name Source: TableSourceScan(table=[[default_catalog, default_database, src]], fields=[id]) exceeded the 80 characters length limit and was truncated.
[IcebergStreamWriter (1/1)#0] INFO org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder - Finished to build heap keyed state-backend.
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode	
[IcebergStreamWriter (1/1)#0] INFO org.apache.flink.runtime.state.heap.HeapKeyedStateBackend - Initializing heap keyed state backend with stream factory.
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.iceberg.BaseMetastoreTableOperations - Refreshing table metadata from new version: file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit13141381595252095998/db.db/test_hash_distribution_mode/metadata/00000-12b9361e-35a0-4862-94ec-9bae7deb0fe1.metadata.json
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.iceberg.BaseMetastoreCatalog - Table loaded by catalog: testhive.db.test_hash_distribution_mode
[IcebergStreamWriter (1/1)#0] INFO org.apache.flink.runtime.taskmanager.Task - IcebergStreamWriter (1/1)#0 (c5292a7cfc1d527e13a5eeb9926a4a95) switched from INITIALIZING to RUNNING.
[Source: TableSourceScan(table=[[default_catalog, default_database, src]], fields=[id]) -> Calc(select=[id, _UTF-16LE'aaa' AS data]) (1/1)#0] INFO org.apache.flink.runtime.taskmanager.Task - Source: TableSourceScan(table=[[default_catalog, default_database, src]], fields=[id]) -> Calc(select=[id, _UTF-16LE'aaa' AS data]) (1/1)#0 (876164887f156d6a29a5e26a541b1d8b) switched from INITIALIZING to RUNNING.
[flink-akka.actor.default-dispatcher-9] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - IcebergStreamWriter (1/1) (c5292a7cfc1d527e13a5eeb9926a4a95) switched from INITIALIZING to RUNNING.
[flink-akka.actor.default-dispatcher-9] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Source: TableSourceScan(table=[[default_catalog, default_database, src]], fields=[id]) -> Calc(select=[id, _UTF-16LE'aaa' AS data]) (1/1) (876164887f156d6a29a5e26a541b1d8b) switched from INITIALIZING to RUNNING.
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.flink.runtime.taskmanager.Task - IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0 (ca05ac4af8adb64ada4a789b93be3848) switched from INITIALIZING to RUNNING.
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1) (ca05ac4af8adb64ada4a789b93be3848) switched from INITIALIZING to RUNNING.
[IcebergStreamWriter (1/1)#0] INFO org.apache.orc.impl.HadoopShimsPre2_7 - Can't get KeyProvider for ORC encryption from hadoop.security.key.provider.path.
[IcebergStreamWriter (1/1)#0] INFO org.apache.orc.impl.PhysicalFsWriter - ORC writer created for path: file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit13141381595252095998/db.db/test_hash_distribution_mode/data/data=aaa/00000-0-99c5c573-3f2b-46df-8a23-e359b36d2c6e-00001.orc with stripeSize: 67108864 blockSize: 268435456 compression: Compress: ZLIB buffer: 262144
[Checkpoint Timer] INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Triggering checkpoint 1 (type=CHECKPOINT) @ 1645512904342 for job 0462ddc2e6a8276f79b3914f9b4fcd19.
[IcebergStreamWriter (1/1)#0] INFO org.apache.orc.impl.WriterImpl - ORC writer created for path: file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit13141381595252095998/db.db/test_hash_distribution_mode/data/data=aaa/00000-0-99c5c573-3f2b-46df-8a23-e359b36d2c6e-00001.orc with stripeSize: 67108864 options: Compress: ZLIB buffer: 262144
[IcebergStreamWriter (1/1)#0] INFO org.apache.iceberg.flink.sink.IcebergStreamWriter - Before checkpoint #1
[IcebergStreamWriter (1/1)#0] INFO org.apache.orc.impl.PhysicalFsWriter - ORC writer created for path: file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit13141381595252095998/db.db/test_hash_distribution_mode/data/data=aaa/00000-0-99c5c573-3f2b-46df-8a23-e359b36d2c6e-00002.orc with stripeSize: 67108864 blockSize: 268435456 compression: Compress: ZLIB buffer: 262144
[IcebergStreamWriter (1/1)#0] INFO org.apache.orc.impl.WriterImpl - ORC writer created for path: file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit13141381595252095998/db.db/test_hash_distribution_mode/data/data=aaa/00000-0-99c5c573-3f2b-46df-8a23-e359b36d2c6e-00002.orc with stripeSize: 67108864 options: Compress: ZLIB buffer: 262144
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.iceberg.flink.sink.IcebergFilesCommitter - Start to flush snapshot state to state backend, table: testhive.db.test_hash_distribution_mode, checkpointId: 1
[jobmanager-io-thread-5] INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed checkpoint 1 for job 0462ddc2e6a8276f79b3914f9b4fcd19 (724151 bytes, checkpointDuration=1418 ms, finalizationTime=6 ms).
[Checkpoint Timer] INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Triggering checkpoint 2 (type=CHECKPOINT) @ 1645512905767 for job 0462ddc2e6a8276f79b3914f9b4fcd19.
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.iceberg.flink.sink.IcebergFilesCommitter - Checkpoint notified, #1
[Source: TableSourceScan(table=[[default_catalog, default_database, src]], fields=[id]) -> Calc(select=[id, _UTF-16LE'aaa' AS data]) (1/1)#0] INFO org.apache.flink.streaming.api.functions.source.datagen.DataGeneratorSource - generated 100000 rows
[Source: TableSourceScan(table=[[default_catalog, default_database, src]], fields=[id]) -> Calc(select=[id, _UTF-16LE'aaa' AS data]) (1/1)#0] INFO org.apache.flink.runtime.taskmanager.Task - Source: TableSourceScan(table=[[default_catalog, default_database, src]], fields=[id]) -> Calc(select=[id, _UTF-16LE'aaa' AS data]) (1/1)#0 (876164887f156d6a29a5e26a541b1d8b) switched from RUNNING to FINISHED.
[Source: TableSourceScan(table=[[default_catalog, default_database, src]], fields=[id]) -> Calc(select=[id, _UTF-16LE'aaa' AS data]) (1/1)#0] INFO org.apache.flink.runtime.taskmanager.Task - Freeing task resources for Source: TableSourceScan(table=[[default_catalog, default_database, src]], fields=[id]) -> Calc(select=[id, _UTF-16LE'aaa' AS data]) (1/1)#0 (876164887f156d6a29a5e26a541b1d8b).
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.taskexecutor.TaskExecutor - Un-registering task and sending final execution state FINISHED to JobManager for task Source: TableSourceScan(table=[[default_catalog, default_database, src]], fields=[id]) -> Calc(select=[id, _UTF-16LE'aaa' AS data]) (1/1)#0 876164887f156d6a29a5e26a541b1d8b.
[flink-akka.actor.default-dispatcher-9] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Source: TableSourceScan(table=[[default_catalog, default_database, src]], fields=[id]) -> Calc(select=[id, _UTF-16LE'aaa' AS data]) (1/1) (876164887f156d6a29a5e26a541b1d8b) switched from RUNNING to FINISHED.
[IcebergStreamWriter (1/1)#0] INFO org.apache.iceberg.flink.sink.IcebergStreamWriter - Before checkpoint #2
[IcebergStreamWriter (1/1)#0] INFO org.apache.orc.impl.PhysicalFsWriter - ORC writer created for path: file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit13141381595252095998/db.db/test_hash_distribution_mode/data/data=aaa/00000-0-99c5c573-3f2b-46df-8a23-e359b36d2c6e-00003.orc with stripeSize: 67108864 blockSize: 268435456 compression: Compress: ZLIB buffer: 262144
[IcebergStreamWriter (1/1)#0] INFO org.apache.orc.impl.WriterImpl - ORC writer created for path: file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit13141381595252095998/db.db/test_hash_distribution_mode/data/data=aaa/00000-0-99c5c573-3f2b-46df-8a23-e359b36d2c6e-00003.orc with stripeSize: 67108864 options: Compress: ZLIB buffer: 262144
[IcebergStreamWriter (1/1)#0] INFO org.apache.iceberg.flink.sink.IcebergStreamWriter - End input reached
[IcebergStreamWriter (1/1)#0] INFO org.apache.flink.runtime.taskmanager.Task - IcebergStreamWriter (1/1)#0 (c5292a7cfc1d527e13a5eeb9926a4a95) switched from RUNNING to FINISHED.
[IcebergStreamWriter (1/1)#0] INFO org.apache.flink.runtime.taskmanager.Task - Freeing task resources for IcebergStreamWriter (1/1)#0 (c5292a7cfc1d527e13a5eeb9926a4a95).
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.taskexecutor.TaskExecutor - Un-registering task and sending final execution state FINISHED to JobManager for task IcebergStreamWriter (1/1)#0 c5292a7cfc1d527e13a5eeb9926a4a95.
[flink-akka.actor.default-dispatcher-9] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - IcebergStreamWriter (1/1) (c5292a7cfc1d527e13a5eeb9926a4a95) switched from RUNNING to FINISHED.
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.iceberg.flink.sink.IcebergFilesCommitter - Committing append with 1 data files and 0 delete files to table testhive.db.test_hash_distribution_mode
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode	
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode	
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 alter_table: db=db tbl=test_hash_distribution_mode newtbl=test_hash_distribution_mode
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 alter_table: db=db tbl=test_hash_distribution_mode newtbl=test_hash_distribution_mode	
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.iceberg.BaseMetastoreTableOperations - Successfully committed to table testhive.db.test_hash_distribution_mode in 193 ms
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.iceberg.SnapshotProducer - Committed snapshot 5883620300433327646 (MergeAppend)
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode	
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.iceberg.BaseMetastoreTableOperations - Refreshing table metadata from new version: file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit13141381595252095998/db.db/test_hash_distribution_mode/metadata/00001-c3fbec69-b9e9-4734-ada1-46b3cf34af00.metadata.json
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode	
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.iceberg.flink.sink.IcebergFilesCommitter - Committed in 378 ms
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.iceberg.flink.sink.IcebergFilesCommitter - Start to flush snapshot state to state backend, table: testhive.db.test_hash_distribution_mode, checkpointId: 2
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.iceberg.flink.sink.IcebergFilesCommitter - End input reached
[jobmanager-io-thread-10] INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed checkpoint 2 for job 0462ddc2e6a8276f79b3914f9b4fcd19 (15055 bytes, checkpointDuration=586 ms, finalizationTime=3 ms).
[Checkpoint Timer] INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Failed to trigger checkpoint for job 0462ddc2e6a8276f79b3914f9b4fcd19 because Some tasks of the job have already finished and checkpointing with finished tasks is not enabled. Failure reason: Not all required tasks are currently running.
[Checkpoint Timer] INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Failed to trigger checkpoint for job 0462ddc2e6a8276f79b3914f9b4fcd19 because Some tasks of the job have already finished and checkpointing with finished tasks is not enabled. Failure reason: Not all required tasks are currently running.
[Checkpoint Timer] INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Failed to trigger checkpoint for job 0462ddc2e6a8276f79b3914f9b4fcd19 because Some tasks of the job have already finished and checkpointing with finished tasks is not enabled. Failure reason: Not all required tasks are currently running.
[Checkpoint Timer] INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Failed to trigger checkpoint for job 0462ddc2e6a8276f79b3914f9b4fcd19 because Some tasks of the job have already finished and checkpointing with finished tasks is not enabled. Failure reason: Not all required tasks are currently running.
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.iceberg.flink.sink.IcebergFilesCommitter - Committing append with 2 data files and 0 delete files to table testhive.db.test_hash_distribution_mode
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode	
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode	
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 alter_table: db=db tbl=test_hash_distribution_mode newtbl=test_hash_distribution_mode
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 alter_table: db=db tbl=test_hash_distribution_mode newtbl=test_hash_distribution_mode	
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.iceberg.BaseMetastoreTableOperations - Successfully committed to table testhive.db.test_hash_distribution_mode in 104 ms
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.iceberg.SnapshotProducer - Committed snapshot 8717152434606897052 (MergeAppend)
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode	
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.iceberg.BaseMetastoreTableOperations - Refreshing table metadata from new version: file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit13141381595252095998/db.db/test_hash_distribution_mode/metadata/00002-3bfc06f1-94da-4443-8dc4-c14e35da3fc0.metadata.json
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode	
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.iceberg.flink.sink.IcebergFilesCommitter - Committed in 258 ms
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.iceberg.flink.sink.IcebergFilesCommitter - Checkpoint notified, #2
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.flink.runtime.taskmanager.Task - IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0 (ca05ac4af8adb64ada4a789b93be3848) switched from RUNNING to FINISHED.
[IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0] INFO org.apache.flink.runtime.taskmanager.Task - Freeing task resources for IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0 (ca05ac4af8adb64ada4a789b93be3848).
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.taskexecutor.TaskExecutor - Un-registering task and sending final execution state FINISHED to JobManager for task IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1)#0 ca05ac4af8adb64ada4a789b93be3848.
[flink-akka.actor.default-dispatcher-8] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - IcebergFilesCommitter -> Sink: IcebergSink testhive.db.test_hash_distribution_mode (1/1) (ca05ac4af8adb64ada4a789b93be3848) switched from RUNNING to FINISHED.
[flink-akka.actor.default-dispatcher-8] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Job insert-into_testhive.db.test_hash_distribution_mode (0462ddc2e6a8276f79b3914f9b4fcd19) switched from state RUNNING to FINISHED.
[flink-akka.actor.default-dispatcher-8] INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Stopping checkpoint coordinator for job 0462ddc2e6a8276f79b3914f9b4fcd19.
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.resourcemanager.slotmanager.DeclarativeSlotManager - Clearing resource requirements of job 0462ddc2e6a8276f79b3914f9b4fcd19
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.dispatcher.StandaloneDispatcher - Job 0462ddc2e6a8276f79b3914f9b4fcd19 reached terminal state FINISHED.
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode	
[flink-akka.actor.default-dispatcher-8] INFO org.apache.flink.runtime.jobmaster.JobMaster - Stopping the JobMaster for job 'insert-into_testhive.db.test_hash_distribution_mode' (0462ddc2e6a8276f79b3914f9b4fcd19).
[flink-akka.actor.default-dispatcher-8] INFO org.apache.flink.runtime.checkpoint.StandaloneCompletedCheckpointStore - Shutting down
[flink-akka.actor.default-dispatcher-8] INFO org.apache.flink.runtime.jobmaster.slotpool.DefaultDeclarativeSlotPool - Releasing slot [55cac133873b326447786e98d4a8bf2b].
[flink-akka.actor.default-dispatcher-8] INFO org.apache.flink.runtime.jobmaster.JobMaster - Close ResourceManager connection c1d6e8672929f08ffdd6d908c456d65a: Stopping JobMaster for job 'insert-into_testhive.db.test_hash_distribution_mode' (0462ddc2e6a8276f79b3914f9b4fcd19).
[flink-akka.actor.default-dispatcher-9] INFO org.apache.flink.runtime.taskexecutor.slot.TaskSlotTableImpl - Free slot TaskSlot(index:0, state:ACTIVE, resource profile: ResourceProfile{taskHeapMemory=256.000gb (274877906944 bytes), taskOffHeapMemory=256.000gb (274877906944 bytes), managedMemory=20.000mb (20971520 bytes), networkMemory=16.000mb (16777216 bytes)}, allocationId: 55cac133873b326447786e98d4a8bf2b, jobId: 0462ddc2e6a8276f79b3914f9b4fcd19).
[flink-akka.actor.default-dispatcher-7] INFO org.apache.flink.runtime.resourcemanager.StandaloneResourceManager - Disconnect job manager a8b41b9177294dfe2cc8799f155e4f8f@akka://flink/user/rpc/jobmanager_6 for job 0462ddc2e6a8276f79b3914f9b4fcd19 from the resource manager.
[flink-akka.actor.default-dispatcher-9] INFO org.apache.flink.runtime.taskexecutor.DefaultJobLeaderService - Remove job 0462ddc2e6a8276f79b3914f9b4fcd19 from job leader monitoring.
[flink-akka.actor.default-dispatcher-9] INFO org.apache.flink.runtime.taskexecutor.TaskExecutor - Close JobManager connection for job 0462ddc2e6a8276f79b3914f9b4fcd19.
[Test worker] INFO org.apache.iceberg.BaseMetastoreTableOperations - Refreshing table metadata from new version: file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit13141381595252095998/db.db/test_hash_distribution_mode/metadata/00002-3bfc06f1-94da-4443-8dc4-c14e35da3fc0.metadata.json
[Test worker] INFO org.apache.iceberg.BaseMetastoreCatalog - Table loaded by catalog: hive.db.test_hash_distribution_mode
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode	
[Test worker] INFO org.apache.iceberg.BaseTableScan - Scanning table hive.db.test_hash_distribution_mode snapshot 5883620300433327646 created at 2022-02-22 06:55:06.089 with filter true
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode	
[Test worker] INFO org.apache.iceberg.BaseMetastoreTableOperations - Refreshing table metadata from new version: file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit13141381595252095998/db.db/test_hash_distribution_mode/metadata/00002-3bfc06f1-94da-4443-8dc4-c14e35da3fc0.metadata.json
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode	
[Test worker] INFO org.apache.iceberg.BaseMetastoreTableOperations - Refreshing table metadata from new version: file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit13141381595252095998/db.db/test_hash_distribution_mode/metadata/00002-3bfc06f1-94da-4443-8dc4-c14e35da3fc0.metadata.json
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_table : db=db tbl=test_hash_distribution_mode	
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 drop_table : db=db tbl=test_hash_distribution_mode
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 drop_table : db=db tbl=test_hash_distribution_mode	
[Test worker] INFO org.apache.iceberg.CatalogUtil - Manifests to delete: GenericManifestFile{path=file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit13141381595252095998/db.db/test_hash_distribution_mode/metadata/46ff29de-4b47-4980-b7c5-9f017f6e1722-m0.avro, length=6101, partition_spec_id=0, added_snapshot_id=8717152434606897052, added_data_files_count=2, added_rows_count=90069, existing_data_files_count=0, existing_rows_count=0, deleted_data_files_count=0, deleted_rows_count=0, partitions=[GenericPartitionFieldSummary{contains_null=false, contains_nan=false, lower_bound=[97, 97, 97], upper_bound=[97, 97, 97]}], key_metadata=null}, GenericManifestFile{path=file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit13141381595252095998/db.db/test_hash_distribution_mode/metadata/eb44b7e6-0d4f-4cd3-b8d9-709e4618e1e3-m0.avro, length=6055, partition_spec_id=0, added_snapshot_id=5883620300433327646, added_data_files_count=1, added_rows_count=9931, existing_data_files_count=0, existing_rows_count=0, deleted_data_files_count=0, deleted_rows_count=0, partitions=[GenericPartitionFieldSummary{contains_null=false, contains_nan=false, lower_bound=[97, 97, 97], upper_bound=[97, 97, 97]}], key_metadata=null}
[Test worker] INFO org.apache.iceberg.hive.HiveCatalog - Dropped table: db.test_hash_distribution_mode
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_table : db=db tbl=test_table
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_table : db=db tbl=test_table	
[Test worker] INFO org.apache.iceberg.BaseMetastoreTableOperations - Refreshing table metadata from new version: file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit13141381595252095998/db.db/test_table/metadata/00000-b94688ab-1fd9-465a-bed0-e4b1a93c98fb.metadata.json
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_table : db=db tbl=test_table
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_table : db=db tbl=test_table	
[Test worker] INFO org.apache.iceberg.BaseMetastoreTableOperations - Refreshing table metadata from new version: file:/var/folders/t6/8l83wbcd52g29cp78v66n7rc0000gn/T/junit13141381595252095998/db.db/test_table/metadata/00000-b94688ab-1fd9-465a-bed0-e4b1a93c98fb.metadata.json
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_table : db=db tbl=test_table
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_table : db=db tbl=test_table	
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 drop_table : db=db tbl=test_table
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 drop_table : db=db tbl=test_table	
[Test worker] INFO org.apache.iceberg.CatalogUtil - Manifests to delete: 
[Test worker] INFO org.apache.iceberg.hive.HiveCatalog - Dropped table: db.test_table
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_database: db
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_database: db	
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 drop_database: db
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 drop_database: db	
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_all_tables: db=db
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_all_tables: db=db	
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore - 1: source:127.0.0.1 get_functions: db=db pat=*
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.HiveMetaStore.audit - ugi=tangyi	ip=127.0.0.1	cmd=source:127.0.0.1 get_functions: db=db pat=*	
[pool-11-thread-1] INFO org.apache.hadoop.hive.metastore.ObjectStore - Dropping database db along with all tables
[Test worker] INFO org.apache.iceberg.hive.HiveCatalog - Dropped namespace: db
[Test worker] ERROR org.apache.iceberg.flink.TestFlinkTableSink - 
--------------------------------------------------------------------------------
Test testHashDistributeMode[catalogName=testhive, baseNamespace=, format=ORC, isStreaming=true](org.apache.iceberg.flink.TestFlinkTableSink) failed with:
java.lang.AssertionError: There should be 1 data file in partition 'aaa' expected:<1> but was:<2>
	at org.junit.Assert.fail(Assert.java:89)
	at org.junit.Assert.failNotEquals(Assert.java:835)
	at org.junit.Assert.assertEquals(Assert.java:647)

[openinx]

@yittg I think the #4189 & #4187 can still fix the all case you described. Would you mind to check & verify this again after apply those two PR ?