Stephen0421 opened a new pull request, #10740: URL: https://github.com/apache/paimon/pull/10740
### Purpose With `sink.coordinator-commit.enabled=true` and `sink.savepoint.auto-tag=true`, stop-with-savepoint on an unaware-bucket append table can finish without creating the savepoint tag. The classic committer path does not have this problem. `CommittingWriteOperatorCoordinator.notifyCheckpointComplete` only queues the commit and `SavepointTagger.tagUpTo` on the single-thread commit executor. For a stop-with-savepoint, Flink closes the coordinator right after the savepoint completes. `close()` called `shutdownNow()`, which drops the queued commit and tag. It also moved to `CLOSED` first, so a queued commit that did start would fail its `RUNNING` check. `close()` now: - drains the commit executor with `ComponentClosingUtils.tryShutdownExecutorElegantly`, waiting at most 30 seconds; - logs a warning if the drain times out; - then moves to `CLOSED` and closes the committer. The timeout stays well below the 60 seconds that `RecreateOnResetOperatorCoordinator` gives `close()` before it interrupts the closing thread and fails the job. Notes: - A commit that is already queued on cancel or global failover now completes instead of being dropped. On global failover, the new coordinator is created only after the old one finishes closing, so recovery can wait up to 30 seconds for that commit. - If the drain times out, the forced shutdown does not wait for an in-flight commit that ignores interrupts before closing the committer. `shutdownNow()` had the same behavior before. - The end-input event handling marked as a TODO in `handleEventFromOperator` is not needed here and is not changed. ### Tests - Re-enabled `AppendTableSavepointTagITCase#testStopWithSavepointCreatesTag` for both the coordinator-commit and classic paths. - Added `CommittingWriteOperatorCoordinatorTest#testCloseDrainsQueuedSavepointTag`. It blocks the commit executor, queues a savepoint commit, and calls `close()`. It fails on the old `close()`. - `CommittingWriteOperatorCoordinatorTest` and `AppendTableSavepointTagITCase` pass with `-Pflink1` on JDK 8 and with `-Pflink2` on JDK 11. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
