JingsongLi commented on code in PR #9309: URL: https://github.com/apache/paimon/pull/9309#discussion_r3817903306
########## paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/AppendTableSavepointTagITCase.java: ########## @@ -0,0 +1,248 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.flink; + +import org.apache.paimon.Snapshot; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.flink.sink.FlinkSinkBuilder; +import org.apache.paimon.flink.sink.SavepointTagUtils; +import org.apache.paimon.flink.source.AbstractNonCoordinatedSource; +import org.apache.paimon.flink.source.AbstractNonCoordinatedSourceReader; +import org.apache.paimon.flink.source.SimpleSourceSplit; +import org.apache.paimon.flink.util.AbstractTestBase; +import org.apache.paimon.table.FileStoreTable; + +import org.apache.flink.api.common.eventtime.WatermarkStrategy; +import org.apache.flink.api.connector.source.Boundedness; +import org.apache.flink.api.connector.source.ReaderOutput; +import org.apache.flink.api.connector.source.SourceReader; +import org.apache.flink.api.connector.source.SourceReaderContext; +import org.apache.flink.core.execution.JobClient; +import org.apache.flink.core.execution.SavepointFormatType; +import org.apache.flink.core.io.InputStatus; +import org.apache.flink.streaming.api.datastream.DataStreamSource; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.api.EnvironmentSettings; +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; + +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * End-to-end savepoint auto-tag tests for unaware-bucket append tables, parameterized over the two + * commit paths (coordinator-commit and the classic global committer). Both must create the same + * {@code savepoint-<checkpointId>} tag for a triggered savepoint. + * + * <p>Uses a source that emits continuously so the async savepoint deterministically lands on a + * data-carrying checkpoint; the empty-savepoint boundary (a separate, shared limitation) is + * intentionally avoided here. + */ +public class AppendTableSavepointTagITCase extends AbstractTestBase { + + // The savepoint tag only materializes once a checkpoint *after* the savepoint completes and + // cumulatively commits the savepoint's snapshot (same catch-up as the classic path). Give the + // poll generous headroom so a transient checkpoint stall under load cannot trip the assertion. + private static final long WAIT_TIMEOUT_MILLIS = 120_000L; + + @ParameterizedTest(name = "coordinatorCommit = {0}") + @ValueSource(booleans = {true, false}) + @Timeout(value = 180, unit = TimeUnit.SECONDS) + public void testSavepointCreatesTag(boolean coordinatorCommit) throws Exception { + String tableName = coordinatorCommit ? "T_COORD" : "T_CLASSIC"; + FileStoreTable table = createTable(tableName, coordinatorCommit); + + JobClient client = runSink(table); + try { + // Wait until a data-carrying snapshot exists so the async savepoint that follows + // deterministically lands on a checkpoint that carries data. + waitUntilSnapshotWithData(table); + + client.triggerSavepoint( + getTempDirPath("savepoint_" + tableName), SavepointFormatType.DEFAULT) + .get(60, TimeUnit.SECONDS); + + // Poll until exactly one savepoint-prefixed tag appears, then assert it is consistent + // with the snapshot it points at. + Map<Snapshot, List<String>> savepointTags = waitUntilSavepointTagCreated(table); + assertThat(savepointTags).hasSize(1); + Map.Entry<Snapshot, List<String>> snapshotWithTags = + savepointTags.entrySet().iterator().next(); + Snapshot tagged = snapshotWithTags.getKey(); + assertThat(snapshotWithTags.getValue()) + .containsExactly(SavepointTagUtils.tagNameOf(tagged.commitIdentifier())); + assertThat(table.snapshotManager().snapshotExists(tagged.id())).isTrue(); + } finally { + client.cancel().get(30, TimeUnit.SECONDS); + } + } + + /** + * A sync savepoint (stop-with-savepoint) receives its own {@code notifyCheckpointComplete}, + * 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). + */ + // TODO: enable once the coordinator supports end-input handling (drains pending tag work). + @Disabled( Review Comment: [P1] Please do not enable this configuration while stop-with-savepoint can lose its tag. `notifyCheckpointComplete` only queues commit/tag work on the coordinator executor, but job termination immediately calls `close()` and `shutdownNow()`, so a successful stop-with-savepoint can finish before `tagUpTo` runs. This test is disabled because it reproduces that exact race. Previously coordinator commit + auto-tag was rejected up front; after this PR users can select it and silently get a savepoint without the required Paimon tag. Please drain/await the queued commit/tag work during termination (or keep the precondition) and enable this test before advertising support. ########## paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/SavepointTagger.java: ########## @@ -0,0 +1,119 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.flink.sink.coordinator; + +import org.apache.paimon.Snapshot; +import org.apache.paimon.flink.sink.SavepointTagUtils; +import org.apache.paimon.operation.TagDeletion; +import org.apache.paimon.table.sink.TagCallback; +import org.apache.paimon.utils.SnapshotManager; +import org.apache.paimon.utils.TagManager; + +import java.io.Serializable; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.NavigableSet; +import java.util.TreeSet; + +import static org.apache.paimon.utils.Preconditions.checkNotNull; + +/** + * Owns savepoint auto-tagging for {@link CommittingWriteOperatorCoordinator}, replicating the + * semantics of the classic {@link + * org.apache.paimon.flink.sink.AutoTagForSavepointCommitterOperator} for the coordinator-commit + * path. It keeps the set of savepoint checkpoint ids still awaiting a snapshot to tag; this set is + * deliberately not checkpointed but rebuilt from the savepoint ids replayed with each subtask's + * committables, so the coordinator's persisted state stays minimal. + */ +public class SavepointTagger { + + private final SnapshotManager snapshotManager; + private final TagManager tagManager; + private final TagDeletion tagDeletion; + private final List<TagCallback> callbacks; + private final Duration tagTimeRetained; + // findSnapshotsForIdentifiers filters by commit user, so the tagger must be bound to the user + // the coordinator actually commits with (which the coordinator restores from its state). + private final String commitUser; + // Checkpoint ids of pending Flink savepoints awaiting a snapshot to tag. + private final NavigableSet<Long> pendingIdentifiers = new TreeSet<>(); + + public SavepointTagger( + SnapshotManager snapshotManager, + TagManager tagManager, + TagDeletion tagDeletion, + List<TagCallback> callbacks, + Duration tagTimeRetained, + String commitUser) { + this.snapshotManager = checkNotNull(snapshotManager); + this.tagManager = checkNotNull(tagManager); + this.tagDeletion = checkNotNull(tagDeletion); + this.callbacks = checkNotNull(callbacks); + this.tagTimeRetained = tagTimeRetained; + this.commitUser = checkNotNull(commitUser); + } + + public void add(long savepointIdentifier) { + pendingIdentifiers.add(savepointIdentifier); + } + + /** + * Tags every pending savepoint whose snapshot the commit up to {@code checkpointId} has + * materialized, then drops those pending intents. + */ + public void tagUpTo(long checkpointId) { + NavigableSet<Long> headSet = pendingIdentifiers.headSet(checkpointId, true); + if (!headSet.isEmpty()) { + createTags(new ArrayList<>(headSet)); + headSet.clear(); + } + } + + /** Drops an aborted savepoint's pending intent and removes any tag already created for it. */ + public void dropAborted(long checkpointId) { + pendingIdentifiers.remove(checkpointId); + deleteTagIfExists(checkpointId); + } + + private void createTags(Collection<Long> identifiers) { + List<Snapshot> snapshots = + snapshotManager.findSnapshotsForIdentifiers( + commitUser, new ArrayList<>(identifiers)); + for (Snapshot snapshot : snapshots) { + String tagName = SavepointTagUtils.tagNameOf(snapshot.commitIdentifier()); + // ignoreIfExists: a later checkpoint's completion may re-tag an already-tagged + // snapshot. + tagManager.createTag(snapshot, tagName, tagTimeRetained, callbacks, true); + } + } + + private void deleteTagIfExists(long id) { + String tagName = SavepointTagUtils.tagNameOf(id); + if (tagManager.tagExists(tagName)) { + tagManager.deleteTag(tagName, tagDeletion, snapshotManager, callbacks); Review Comment: [P2] Only delete a tag when this coordinator observed/created the matching savepoint intent. `notifyCheckpointAborted` is invoked for normal checkpoints too, but `dropAborted` ignores whether the id was pending and unconditionally deletes the global `savepoint-<checkpointId>` name. Checkpoint ids restart in a fresh job, so aborting normal checkpoint N can delete a valid `savepoint-N` retained from an earlier job (and tag deletion may release its protected files). Please track ownership/created ids, or validate the tag snapshot/commit user before deleting it; add a test where a same-named pre-existing tag exists without a pending intent. -- 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]
