You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by yu...@apache.org on 2026/07/22 09:53:47 UTC
(fluss) branch main updated: [tiering] Use exponential delay restart strategy for Flink tiering job (#3683)
This is an automated email from the ASF dual-hosted git repository.
luoyuxia pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new f2fa4f814 [tiering] Use exponential delay restart strategy for Flink tiering job (#3683)
f2fa4f814 is described below
commit f2fa4f814e1f3cbee94eea1bcc5331a0a6dff0e4
Author: Shawn Huang <hx...@gmail.com>
AuthorDate: Wed Jul 22 17:53:42 2026 +0800
[tiering] Use exponential delay restart strategy for Flink tiering job (#3683)
---
.../fluss/flink/tiering/FlinkTieringTestBase.java | 1 -
fluss-flink/fluss-flink-tiering/pom.xml | 73 +++++++++++++++++++++-
.../fluss/flink/tiering/FlussLakeTiering.java | 7 +++
.../fluss/flink/tiering/TieringFailoverITCase.java | 30 ++++++++-
4 files changed, 106 insertions(+), 5 deletions(-)
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/FlinkTieringTestBase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/FlinkTieringTestBase.java
index 7e3ee48a7..aab355718 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/FlinkTieringTestBase.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/FlinkTieringTestBase.java
@@ -103,7 +103,6 @@ class FlinkTieringTestBase {
execEnv = StreamExecutionEnvironment.getExecutionEnvironment();
execEnv.setRuntimeMode(RuntimeExecutionMode.STREAMING);
execEnv.setParallelism(2);
- execEnv.enableCheckpointing(500);
}
protected long createTable(TablePath tablePath, Schema schema) throws Exception {
diff --git a/fluss-flink/fluss-flink-tiering/pom.xml b/fluss-flink/fluss-flink-tiering/pom.xml
index 45917b886..e504246a4 100644
--- a/fluss-flink/fluss-flink-tiering/pom.xml
+++ b/fluss-flink/fluss-flink-tiering/pom.xml
@@ -57,6 +57,77 @@
<version>${flink.minor.version}</version>
<scope>provided</scope>
</dependency>
+
+ <!-- test dependencies -->
+ <dependency>
+ <groupId>org.apache.fluss</groupId>
+ <artifactId>fluss-flink-common</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.fluss</groupId>
+ <artifactId>fluss-server</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.fluss</groupId>
+ <artifactId>fluss-server</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.fluss</groupId>
+ <artifactId>fluss-common</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.fluss</groupId>
+ <artifactId>fluss-test-utils</artifactId>
+ </dependency>
+
+ <!-- for curator TestingServer -->
+ <dependency>
+ <groupId>org.apache.curator</groupId>
+ <artifactId>curator-test</artifactId>
+ <version>${curator.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-clients</artifactId>
+ <version>${flink.minor.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-base</artifactId>
+ <version>${flink.minor.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <!--
+ Required at test runtime because TieringSource loads FlinkConnectorOptionsUtils, which
+ references ValidationException from flink-table-common. fluss-flink-common declares this
+ dependency as provided, so it is not transitively available to this module's tests.
+ -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-common</artifactId>
+ <version>${flink.minor.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
@@ -85,4 +156,4 @@
</plugins>
</build>
-</project>
\ No newline at end of file
+</project>
diff --git a/fluss-flink/fluss-flink-tiering/src/main/java/org/apache/fluss/flink/tiering/FlussLakeTiering.java b/fluss-flink/fluss-flink-tiering/src/main/java/org/apache/fluss/flink/tiering/FlussLakeTiering.java
index cead3a035..ba498e513 100644
--- a/fluss-flink/fluss-flink-tiering/src/main/java/org/apache/fluss/flink/tiering/FlussLakeTiering.java
+++ b/fluss-flink/fluss-flink-tiering/src/main/java/org/apache/fluss/flink/tiering/FlussLakeTiering.java
@@ -24,6 +24,7 @@ import org.apache.fluss.config.provider.ConfigProviders;
import org.apache.fluss.flink.adapter.MultipleParameterToolAdapter;
import org.apache.flink.configuration.JobManagerOptions;
+import org.apache.flink.configuration.RestartStrategyOptions;
import org.apache.flink.core.execution.JobClient;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -31,6 +32,7 @@ import java.util.HashMap;
import java.util.Map;
import java.util.ServiceLoader;
+import static org.apache.flink.configuration.RestartStrategyOptions.RestartStrategyType.EXPONENTIAL_DELAY;
import static org.apache.flink.runtime.executiongraph.failover.FailoverStrategyFactoryLoader.FULL_RESTART_STRATEGY_NAME;
import static org.apache.fluss.flink.tiering.source.TieringSourceOptions.DATA_LAKE_CONFIG_PREFIX;
import static org.apache.fluss.utils.PropertiesUtils.extractAndRemovePrefix;
@@ -98,6 +100,11 @@ public class FlussLakeTiering {
org.apache.flink.configuration.Configuration flinkConfig =
new org.apache.flink.configuration.Configuration();
flinkConfig.set(JobManagerOptions.EXECUTION_FAILOVER_STRATEGY, FULL_RESTART_STRATEGY_NAME);
+ // Configure restart strategy: the tiering job is completely stateless (offsets are
+ // 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, EXPONENTIAL_DELAY.getMainValue());
execEnv = StreamExecutionEnvironment.getExecutionEnvironment(flinkConfig);
}
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/TieringFailoverITCase.java b/fluss-flink/fluss-flink-tiering/src/test/java/org/apache/fluss/flink/tiering/TieringFailoverITCase.java
similarity index 79%
rename from fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/TieringFailoverITCase.java
rename to fluss-flink/fluss-flink-tiering/src/test/java/org/apache/fluss/flink/tiering/TieringFailoverITCase.java
index 07b2ddbd0..0f4d1a610 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/TieringFailoverITCase.java
+++ b/fluss-flink/fluss-flink-tiering/src/test/java/org/apache/fluss/flink/tiering/TieringFailoverITCase.java
@@ -18,16 +18,20 @@
package org.apache.fluss.flink.tiering;
+import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.lake.values.TestingValuesLake;
+import org.apache.fluss.metadata.DataLakeFormat;
import org.apache.fluss.metadata.Schema;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.row.InternalRow;
import org.apache.fluss.types.DataTypes;
+import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.core.execution.JobClient;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.util.ArrayList;
@@ -59,6 +63,24 @@ class TieringFailoverITCase extends FlinkTieringTestBase {
FlinkTieringTestBase.afterAll();
}
+ @BeforeEach
+ @Override
+ void beforeEach() {
+ String bootstrapServers = String.join(",", clientConf.get(ConfigOptions.BOOTSTRAP_SERVERS));
+ String[] args = {
+ "--fluss.bootstrap.servers",
+ bootstrapServers,
+ "--datalake.format",
+ DataLakeFormat.PAIMON.toString()
+ };
+ FlussLakeTiering tiering = new FlussLakeTiering(args);
+ execEnv = tiering.execEnv;
+ execEnv.setRuntimeMode(RuntimeExecutionMode.STREAMING);
+ execEnv.setParallelism(2);
+
+ assertThat(execEnv.getCheckpointConfig().isCheckpointingEnabled()).isFalse();
+ }
+
@Test
void testTiering() throws Exception {
// create a pk table, write some records and wait until snapshot finished
@@ -71,8 +93,8 @@ class TieringFailoverITCase extends FlinkTieringTestBase {
writeRows(t1, rows, false);
FLUSS_CLUSTER_EXTENSION.triggerAndWaitSnapshot(t1);
- // fail the first write to the pk table
- TestingValuesLake.failWhen(t1.toString()).failWriteOnce();
+ // fail the first two writes to the pk table
+ TestingValuesLake.failWhen(t1.toString()).failWriteNext(2);
// then start tiering job
JobClient jobClient = buildTieringJob(execEnv);
@@ -96,7 +118,9 @@ class TieringFailoverITCase extends FlinkTieringTestBase {
checkDataInValuesTable(t1, expectedRows);
} finally {
- jobClient.cancel().get();
+ if (!jobClient.getJobExecutionResult().isDone()) {
+ jobClient.cancel().get();
+ }
}
}