When distribution-mode = RANGE, Flink Stop-with-savepoint operation will definitely fail

[huyuanfeng2018]

Apache Iceberg version

1.7.0 (latest release)

Query engine

Flink

Please describe the bug 🐞

When my job turns on distribution-mode = RANGE, and I try to stop the task and generate Savepoint at the same time, an exception will occur.

The key case is:

The vertex Source:xxx[10] -> Calc[11] -> range-shuffle (id = xxxxx) has used UnionListState, but part of its tasks has called operators' finish method.

Willingness to contribute

• I can contribute a fix for this bug independently
• I would be willing to contribute a fix for this bug with guidance from the Iceberg community
• I cannot contribute a fix for this bug at this time