[Bug] when Dedicated Compaction use combined mode, restore from checkpoint ,then error：This exception is intentionally thrown after committing the restored checkpoints By restarting the job we hope that writers can start writing based on these new commits.

[dsanww]

Search before asking

• I searched in the issues and found nothing similar.

Paimon version

paimon-flink-action-1.0-20240923.002611-17.jar

Compute Engine

flink1.18

Minimal reproduce step

when Dedicated Compaction use combined mode, restore from checkpoint ,then error：This exception is intentionally thrown after committing the restored checkpoints By restarting the job we hope that writers can start writing based on these new commits.

error log:
2024-09-26 16:33:42,289 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Global Committer (1/2) (63a4f167eb1b7dda7f3f0553fdd3dbd2_4a24f7c06caba60191974000475ccc9e_0_2) switched from INITIALIZING to FAILED on container_e46_1672229612061_4391574_01_000002 @ emr-worker-13.cluster-258643 (dataPort=33915).
java.lang.RuntimeException: This exception is intentionally thrown after committing the restored checkpoints. By restarting the job we hope that writers can start writing based on these new commits.
at org.apache.paimon.flink.sink.RestoreAndFailCommittableStateManager.recover(RestoreAndFailCommittableStateManager.java:84) ~[paimon-flink-1.18-1.0-20240923.002611-17.jar:1.0-SNAPSHOT]
at org.apache.paimon.flink.sink.RestoreAndFailCommittableStateManager.initializeState(RestoreAndFailCommittableStateManager.java:77) ~[paimon-flink-1.18-1.0-20240923.002611-17.jar:1.0-SNAPSHOT]
at org.apache.paimon.flink.sink.CommitterOperator.initializeState(CommitterOperator.java:153) ~[paimon-flink-1.18-1.0-20240923.002611-17.jar:1.0-SNAPSHOT]
at org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.initializeOperatorState(StreamOperatorStateHandler.java:122) ~[flink-dist-1.18.0.jar:1.18.0]
at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:274) ~[flink-dist-1.18.0.jar:1.18.0]
at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.initializeStateAndOpenOperators(RegularOperatorChain.java:106) ~[flink-dist-1.18.0.jar:1.18.0]
at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreGates(StreamTask.java:753) ~[flink-dist-1.18.0.jar:1.18.0]
at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55) ~[flink-dist-1.18.0.jar:1.18.0]
at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreInternal(StreamTask.java:728) ~[flink-dist-1.18.0.jar:1.18.0]
at org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:693) ~[flink-dist-1.18.0.jar:1.18.0]
at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:953) ~[flink-dist-1.18.0.jar:1.18.0]
at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:922) ~[flink-dist-1.18.0.jar:1.18.0]
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:746) ~[flink-dist-1.18.0.jar:1.18.0]
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:562) ~[flink-dist-1.18.0.jar:1.18.0]
at java.lang.Thread.run(Thread.java:748) ~[?:1.8.0_252]

What doesn't meet your expectations?

error

Anything else?

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!

[Tan-JiaLiang]

I think org.apache.paimon.flink.sink.CombinedTableCompactorSink#createCommittableStateManager should return org.apache.paimon.flink.sink.NoopCommittableStateManager. WDYT? cc @leaves12138 @tsreaper