[minor][lake] Set default name for flink tiering service job (apache#1105)

Author: leonardBang

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

@@ -28,6 +28,7 @@
 import com.alibaba.fluss.lake.writer.LakeTieringFactory;
 
 import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.configuration.PipelineOptions;
 import org.apache.flink.core.execution.JobClient;
 import org.apache.flink.streaming.api.datastream.DataStreamSource;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -42,6 +43,8 @@
 /** The builder to build Flink lake tiering job. */
 public class LakeTieringJobBuilder {
 
+    private static final String DEFAULT_TIERING_SERVICE_JOB_NAME = "Fluss Lake Tiering Service";
+
     private final StreamExecutionEnvironment env;
     private final Configuration flussConfig;
     private final Configuration dataLakeConfig;
@@ -106,7 +109,11 @@ public JobClient build() throws Exception {
                 .setParallelism(1)
                 .setMaxParallelism(1)
                 .sinkTo(new DiscardingSink());
+        String jobName =
+                env.getConfiguration()
+                        .getOptional(PipelineOptions.NAME)
+                        .orElse(DEFAULT_TIERING_SERVICE_JOB_NAME);
 
-        return env.executeAsync();
+        return env.executeAsync(jobName);
     }
 }

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

@@ -30,7 +30,7 @@
 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. */
+/** The entrypoint for Flink to tier fluss data to lake format like paimon. */
 public class FlussLakeTieringEntrypoint {
 
     private static final String FLUSS_CONF_PREFIX = "fluss.";