[huyuanfeng2018]

Purpose

Fixes #11138.

The writer coordinator refreshes its cached metadata when a checkpoint is triggered, before the committer publishes that checkpoint's table snapshot. The cache survives task recovery, so a restored primary-key writer can initialize its sequence numbers from older files and silently lose replayed updates to already committed rows.

Invalidate the metadata view on both global and subtask recovery, then refresh the snapshot and cached commit identifiers on the existing request executor before serving the next request. Track recovery generations so a failed refresh is retried and a reset during an in-flight refresh is not lost. Keep existing paged responses available to unaffected subtasks.

Tests

• Reproduced on unmodified upstream master 57233dbc18db4cefa45ab1447a44843368b1191e: both recovery metadata regressions fail, and the Flink 1.20.1 MiniCluster test returns version 4 instead of 5 with the coordinator enabled. The coordinator-disabled control passes.
• The MiniCluster regression uses a checkpointed replayable source, manually triggered checkpoints, an injected task failure, and a real Parquet primary-key table. It verifies the restored source position and the final committed row.
• Unit coverage includes global/subtask recovery, reset before start, commit identifier refresh, failed refresh retry, a second reset during refresh, and continued consumption of in-flight paged responses.

Validation on Temurin JDK 8u512 and Flink 1.20.1:

• WriteOperatorCoordinatorTest: 7 passed.
• TableWriteCoordinatorTest: 7 passed.
• WriterCoordinatorRecoveryITCase: 2 passed (coordinator enabled/disabled).
• Checkstyle, Spotless, Maven enforcer checks, and git diff --check passed.

mvn -pl paimon-flink/paimon-flink-common -Pflink1 \
  -DwildcardSuites=none -Dflink.forkCount=1 \
  -Dtest=WriteOperatorCoordinatorTest,TableWriteCoordinatorTest,WriterCoordinatorRecoveryITCase test