ifndef-SleePy commented on code in PR #9309: URL: https://github.com/apache/paimon/pull/9309#discussion_r3827977344
########## 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: Yes, this is a real issue. I looked into using a drain as a lightweight fix for stop-with-savepoint when auto-tagging is enabled with coordinator commit, before end-input support is available. Draining the executor in `CommittingWriteOperatorCoordinator.close()` should work in the normal case. **However, if processing the final savepoint completion fails to create the tag, it cannot provide the same guarantee as the operator path.** 1. For coordinator commit, `SchedulerBase.closeAsync` disposes operator coordinators through `IOUtils.closeQuietly`, which ignores exceptions from `close()`. A failed drain could therefore leave the savepoint successful and the job finished, while the tag was never created. 2. For operator commit, `StreamTask.afterInvoke` waits for `finalCheckpointCompleted` before finishing. If the final `notifyCheckpointComplete` fails, the task fails. The savepoint is still created, but the job ends in `FAILED` state: `StopWithSavepointTerminationHandlerImpl` calls `scheduler.handleGlobalFailure` when a task does not finish, and `StopWithSavepointStoppingException` is a `NonRecoverableError`. The stop-with-savepoint operation then fails visibly. As a result, the two paths behave differently when creating the tag for the final savepoint fails: | | Savepoint | Job | Tag | Stop-with-savepoint result | | --- | --- | --- | --- | --- | | Coordinator commit | Successful | `FINISHED` | Not created | Completes successfully | | Operator commit | Successful | `FAILED` | Not created | Completes exceptionally | This difference may be confusing for users. I am working on a complete solution. The PoC validates the basic path, but completing and verifying consistent behavior across failover scenarios will take additional time. It may require synchronizing the task and coordinator so the final tag work is drained before termination. If the change grows beyond the scope of this PR, I may address it in a separate PR. -- 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]
