This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 43892581de [flink] Drain the commit coordinator on close so
stop-with-savepoint … (#10740)
43892581de is described below
commit 43892581de1cb39a9fba039e9f5444c1aeb69859
Author: Wenchao Wu <[email protected]>
AuthorDate: Sat Oct 24 09:42:39 2026 +0800
[flink] Drain the commit coordinator on close so stop-with-savepoint …
(#10740)
---
.../CommittingWriteOperatorCoordinator.java | 23 ++++++++++--
.../flink/AppendTableSavepointTagITCase.java | 16 ++-------
.../CommittingWriteOperatorCoordinatorTest.java | 41 ++++++++++++++++++++++
3 files changed, 64 insertions(+), 16 deletions(-)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CommittingWriteOperatorCoordinator.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CommittingWriteOperatorCoordinator.java
index e792432a31..7df3678965 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CommittingWriteOperatorCoordinator.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CommittingWriteOperatorCoordinator.java
@@ -32,6 +32,7 @@ import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.core.io.SimpleVersionedSerialization;
import org.apache.flink.core.io.SimpleVersionedSerializerTypeSerializerProxy;
import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.operators.coordination.ComponentClosingUtils;
import org.apache.flink.runtime.operators.coordination.OperatorCoordinator;
import org.apache.flink.runtime.operators.coordination.OperatorEvent;
import
org.apache.flink.runtime.operators.coordination.RecreateOnResetOperatorCoordinator;
@@ -42,6 +43,7 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
import java.io.IOException;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -75,6 +77,10 @@ public class CommittingWriteOperatorCoordinator implements
OperatorCoordinator {
private static final Logger LOG =
LoggerFactory.getLogger(CommittingWriteOperatorCoordinator.class);
+ // Must stay well below the 60s that RecreateOnResetOperatorCoordinator
gives close(), after
+ // which Flink interrupts the closing thread and fails the job.
+ private static final Duration DRAIN_TIMEOUT = Duration.ofSeconds(30);
+
private final OperatorCoordinator.Context context;
private final Committer.Factory<Committable, ManifestCommittable>
committerFactory;
private final boolean streamingCheckpointEnabled;
@@ -154,10 +160,16 @@ public class CommittingWriteOperatorCoordinator
implements OperatorCoordinator {
@Override
public void close() throws Exception {
- transitionState(State.CLOSED);
- if (commitExecutor != null) {
- commitExecutor.shutdownNow();
+ // Let actions already queued run before closing, e.g. the commit and
savepoint tag queued
+ // by the notifyCheckpointComplete of a stop-with-savepoint right
before the job finishes.
+ // They require RUNNING, so the state changes only after draining.
+ if
(!ComponentClosingUtils.tryShutdownExecutorElegantly(commitExecutor,
DRAIN_TIMEOUT)) {
+ LOG.warn(
+ "Commit executor did not finish queued actions within {}.
A pending commit or"
+ + " savepoint tag may be lost.",
+ DRAIN_TIMEOUT);
}
+ transitionState(State.CLOSED);
if (committer != null) {
committer.close();
committer = null;
@@ -602,6 +614,11 @@ public class CommittingWriteOperatorCoordinator implements
OperatorCoordinator {
return state;
}
+ @VisibleForTesting
+ boolean isCommitExecutorShutdown() {
+ return commitExecutor.isShutdown();
+ }
+
@VisibleForTesting
String getCommitUser() {
return commitUser;
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/AppendTableSavepointTagITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/AppendTableSavepointTagITCase.java
index 759f5567bf..caa91d5bcb 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/AppendTableSavepointTagITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/AppendTableSavepointTagITCase.java
@@ -43,7 +43,6 @@ import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
-import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
@@ -107,19 +106,10 @@ public class AppendTableSavepointTagITCase extends
AbstractTestBase {
* unlike an async savepoint, so the tag is created for the savepoint's
own snapshot rather than
* caught up by a later checkpoint. Both commit paths must still produce
the same tag.
*
- * <p>Disabled for now: on the coordinator-commit path this is racy. The
coordinator creates the
- * tag asynchronously on its single-thread commit executor
(notifyCheckpointComplete ->
- * tagUpTo), but stop-with-savepoint terminates the job right after the
savepoint, and the
- * coordinator's {@code close()} calls {@code
commitExecutor.shutdownNow()}, which can drop the
- * not-yet-run tag task so the tag is silently lost. The classic operator
path is unaffected
- * because it tags synchronously. Re-enable once the coordinator drains
pending commit/tag work
- * on end-of-input shutdown (the follow-up PR that adds proper end-input
handling to the
- * coordinator).
+ * <p>On the coordinator-commit path the tag is created on the commit
executor, which the
+ * coordinator drains on close. That can happen after the
stop-with-savepoint future returns, so
+ * the tag must be polled for.
*/
- // TODO: enable once the coordinator supports end-input handling (drains
pending tag work).
- @Disabled(
- "Coordinator-commit stop-with-savepoint drops the async tag on
shutdownNow; re-enable"
- + " after the coordinator end-input handling PR drains
pending tag work")
@ParameterizedTest(name = "coordinatorCommit = {0}")
@ValueSource(booleans = {true, false})
@Timeout(value = 180, unit = TimeUnit.SECONDS)
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/CommittingWriteOperatorCoordinatorTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/CommittingWriteOperatorCoordinatorTest.java
index 2357eaccc4..57ac6b66aa 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/CommittingWriteOperatorCoordinatorTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/CommittingWriteOperatorCoordinatorTest.java
@@ -69,6 +69,7 @@ import java.util.Map;
import java.util.NavigableMap;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
@@ -481,6 +482,46 @@ public class CommittingWriteOperatorCoordinatorTest
extends CommitterTestBase {
coordinator.close();
}
+ @Timeout(value = 30, unit = TimeUnit.SECONDS)
+ @Test
+ public void testCloseDrainsQueuedSavepointTag() throws Exception {
+ FileStoreTable table = createUnawareBucketTable();
+ TestingContext context = new TestingContext(new OperatorID(), 1);
+ CommittingWriteOperatorCoordinator coordinator =
+ createCoordinatorWithAutoTag(table, context);
+ coordinator.start();
+ coordinator.waitProcessAllActions();
+
+ // Hold the commit executor so the commit and tag of a
stop-with-savepoint are still
+ // queued when close() starts.
+ CountDownLatch release = new CountDownLatch(1);
+ coordinator.runInEventLoop(release::await, "blocking");
+ coordinator.handleEventFromOperator(
+ 0, 0, savepointEvent(1L,
Collections.singletonList(committable(table, 1, 1))));
+ coordinator.notifyCheckpointComplete(1L);
+
+ CompletableFuture<Void> closed =
+ CompletableFuture.runAsync(
+ () -> {
+ try {
+ coordinator.close();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+ // Release the executor only once close() has shut it down.
+ while (!coordinator.isCommitExecutorShutdown()) {
+ Thread.sleep(10);
+ }
+ release.countDown();
+ closed.get();
+
+ assertThat(coordinator.getCurrentState())
+ .isEqualTo(CommittingWriteOperatorCoordinator.State.CLOSED);
+
assertThat(table.snapshotManager().latestSnapshot().commitIdentifier()).isEqualTo(1L);
+ assertThat(table.tagManager().tagExists(savepointTag(1L))).isTrue();
+ }
+
private static String savepointTag(long checkpointId) {
return SavepointTagUtils.tagNameOf(checkpointId);
}
