[tiering] Use fixed-delay restart strategy for Flink tiering job#3683
[tiering] Use fixed-delay restart strategy for Flink tiering job#3683Shawn-Hx wants to merge 1 commit into
Conversation
Configure the stateless tiering job to retry indefinitely when checkpointing is disabled, preventing transient failures from terminating the job permanently. Move the failover integration test to fluss-flink-tiering so that it covers the production job configuration.
There was a problem hiding this comment.
Pull request overview
This PR updates the Flink tiering job entrypoint to use a fixed-delay restart strategy (to avoid Flink’s default no-restart behavior when checkpointing is disabled) and adjusts tiering failover integration testing so it exercises the production entrypoint’s StreamExecutionEnvironment configuration.
Changes:
- Configure
FlussLakeTieringto use Flink’s fixed-delay restart strategy with effectively unlimited restart attempts. - Update
TieringFailoverITCaseto build itsStreamExecutionEnvironmentviaFlussLakeTiering, inject two consecutive lake write failures, and avoid canceling an already-finished job. - Remove checkpoint enabling from the shared tiering IT base and add needed test dependencies to the tiering module.
Reviewed changes
Copilot reviewed 4 out of 4 changed files in this pull request and generated 1 comment.
| File | Description |
|---|---|
| fluss-flink/fluss-flink-tiering/src/test/java/org/apache/fluss/flink/tiering/TieringFailoverITCase.java | Reworks failover IT to use the production entrypoint’s exec env and validates recovery from consecutive write failures. |
| fluss-flink/fluss-flink-tiering/src/main/java/org/apache/fluss/flink/tiering/FlussLakeTiering.java | Adds fixed-delay restart strategy configuration for stateless tiering jobs when checkpoints are disabled. |
| fluss-flink/fluss-flink-tiering/pom.xml | Adds test-scoped dependencies needed to run tiering ITs in the tiering module. |
| fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/FlinkTieringTestBase.java | Aligns tiering IT base with production by no longer enabling checkpointing by default. |
Comments suppressed due to low confidence (1)
fluss-flink/fluss-flink-tiering/src/test/java/org/apache/fluss/flink/tiering/TieringFailoverITCase.java:75
- This test bootstraps the ExecutionEnvironment via FlussLakeTiering, but passes
--datalake.formatas Paimon while the actual tiering job built in FlinkTieringTestBase is hard-coded to useDataLakeFormat.LANCE.toString()(see FlinkTieringTestBase#buildTieringJob). Using LANCE here keeps the test configuration consistent and avoids masking format-specific behavior in FlussLakeTiering initialization.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| flinkConfig.set(RestartStrategyOptions.RESTART_STRATEGY, "fixed-delay"); | ||
| flinkConfig.set( | ||
| RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_ATTEMPTS, Integer.MAX_VALUE); |
| // persisted in Fluss Server's LakeSnapshot, not in Flink checkpoint state), so it is | ||
| // safe to restart indefinitely on failure. Without this, Flink defaults to "no-restart" | ||
| // when checkpoint is not enabled, causing the job to fail permanently on any exception. | ||
| flinkConfig.set(RestartStrategyOptions.RESTART_STRATEGY, "fixed-delay"); |
There was a problem hiding this comment.
Before this change, the failover test enabled checkpointing and therefore inherited Flink 1.20’s default exponential-delay restart policy (1s initial backoff, 1.5 multiplier, and 1min maximum backoff). Selecting fixed-delay here changes the behavior to retry every 1s indefinitely. During a persistent lake/storage failure, this can continuously hit the external system and flood logs instead of backing off. Is this fixed retry cadence intentional? If the goal is to retain the retry behavior previously exercised by the checkpoint-enabled test while enabling restarts without checkpoints, could we explicitly select exponential-delay instead (whose default attempt count is effectively infinite), or otherwise document and explicitly configure the fixed-delay policy?
| <scope>test</scope> | ||
| </dependency> | ||
|
|
||
| <dependency> |
There was a problem hiding this comment.
Is this dependency needed for the moved failover IT? I could not find any org.apache.flink.table usage in TieringFailoverITCase, FlinkTieringTestBase, or the tiering execution path exercised by this test. The other newly added dependencies map to concrete test classes or runtime requirements, while this one appears unnecessary. Could we remove it to keep the test classpath minimal, or document which class requires it?
Configure the stateless tiering job to retry indefinitely when checkpointing is disabled, preventing transient failures from terminating the job permanently.
Move the failover integration test to fluss-flink-tiering so that it covers the production job configuration.
Purpose
Linked issue: close #3681
Brief change log
Tests
TieringFailoverITCaseAPI and Format
No public API or storage format changes.
Documentation
No documentation changes are required.