[hotfix] Always use full restart strategy for tiering service to avoid table can't be committed successfully (apache#1103)

Author: luoyuxia

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/tiering/source/TieringSourceReader.java

@@ -26,6 +26,8 @@
 import org.apache.flink.api.connector.source.SourceReaderContext;
 import org.apache.flink.connector.base.source.reader.SingleThreadMultiplexSourceReaderBase;
 
+import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 
 /** A {@link SourceReader} that read records from Fluss and write to lake. */
@@ -61,6 +63,12 @@ protected void onSplitFinished(Map<String, TieringSplitState> finishedSplitIds)
         context.sendSplitRequest();
     }
 
+    @Override
+    public List<TieringSplit> snapshotState(long checkpointId) {
+        // we return empty list to make source reader be stateless
+        return Collections.emptyList();
+    }
+
     @Override
     protected TieringSplitState initializedState(TieringSplit split) {
         if (split.isTieringSnapshotSplit()) {

fluss-flink/fluss-flink-tiering/src/main/java/com/alibaba/fluss/flink/tiering/FlussLakeTieringEntrypoint.java

@@ -20,13 +20,15 @@
 import com.alibaba.fluss.config.Configuration;
 
 import org.apache.flink.api.java.utils.MultipleParameterTool;
+import org.apache.flink.configuration.JobManagerOptions;
 import org.apache.flink.core.execution.JobClient;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
 import java.util.Map;
 
 import static com.alibaba.fluss.flink.tiering.source.TieringSourceOptions.DATA_LAKE_CONFIG_PREFIX;
 import static com.alibaba.fluss.utils.PropertiesUtils.extractAndRemovePrefix;
+import static org.apache.flink.runtime.executiongraph.failover.FailoverStrategyFactoryLoader.FULL_RESTART_STRATEGY_NAME;
 
 /** The entrypoint for Flink to tiering fluss data to Paimon. */
 public class FlussLakeTieringEntrypoint {
@@ -62,9 +64,17 @@ public static void main(String[] args) throws Exception {
                 extractAndRemovePrefix(
                         paramsMap, String.format("%s%s.", DATA_LAKE_CONFIG_PREFIX, dataLake));
 
+        // now, we must use full restart strategy if any task is failed,
+        // since committer is stateless, if tiering committer is failover, committer
+        // will lost the collected committable, and will never collect all committable to do commit
+        // todo: support region failover
+        org.apache.flink.configuration.Configuration flinkConfig =
+                new org.apache.flink.configuration.Configuration();
+        flinkConfig.set(JobManagerOptions.EXECUTION_FAILOVER_STRATEGY, FULL_RESTART_STRATEGY_NAME);
+
         // build tiering source
         final StreamExecutionEnvironment execEnv =
-                StreamExecutionEnvironment.getExecutionEnvironment();
+                StreamExecutionEnvironment.getExecutionEnvironment(flinkConfig);
 
         // build lake tiering job
         JobClient jobClient =