[Fix-17908][Flink] Remove -sae parameter for APPLICATION mode to prevent task termination

The -sae (--shutdownOnAttachedExit) parameter is only suitable for attached mode
where the CLI stays connected and waits for the job to complete. However,
YARN Application mode runs in detached mode where the CLI exits after
submission, causing the -sae parameter to trigger cluster shutdown and
terminate the job unexpectedly.

Changes:
- Only add -sae parameter for non-APPLICATION deploy modes
- Update test cases to reflect the new behavior

Author: chris-fast

dolphinscheduler-task-plugin/dolphinscheduler-task-flink-stream/src/test/java/org/apache/dolphinscheduler/plugin/task/flink/FlinkArgsUtilsTest.java

@@ -68,8 +68,9 @@ public void testRunJarInApplicationMode() throws Exception {
         FlinkStreamParameters flinkParameters = buildTestFlinkParametersWithDeployMode(FlinkDeployMode.APPLICATION);
         List<String> commandLine = FlinkArgsUtils.buildRunCommandLine(buildTestTaskExecutionContext(), flinkParameters);
 
+        // APPLICATION mode should NOT include -sae parameter (detached mode on YARN)
         Assertions.assertEquals(
-                "${FLINK_HOME}/bin/flink run-application -t yarn-application -ys 4 -ynm demo-app-name -yjm 1024m -ytm 1024m -p 4 -sae -c org.example.Main /opt/job.jar",
+                "${FLINK_HOME}/bin/flink run-application -t yarn-application -ys 4 -ynm demo-app-name -yjm 1024m -ytm 1024m -p 4 -c org.example.Main /opt/job.jar",
                 joinStringListWithSpace(commandLine));
     }
 

dolphinscheduler-task-plugin/dolphinscheduler-task-flink/src/main/java/org/apache/dolphinscheduler/plugin/task/flink/FlinkArgsUtils.java

@@ -267,7 +267,10 @@ private static List<String> buildRunCommandLineForOthers(TaskExecutionContext ta
         // If the job is submitted in attached mode, perform a best-effort cluster shutdown when the CLI is terminated
         // abruptly
         // The task status will be synchronized with the cluster job status
-        args.add(FlinkConstants.FLINK_SHUTDOWN_ON_ATTACHED_EXIT); // -sae
+        // Note: -sae should NOT be used for APPLICATION mode, as it runs in detached mode on YARN
+        if (deployMode != FlinkDeployMode.APPLICATION) {
+            args.add(FlinkConstants.FLINK_SHUTDOWN_ON_ATTACHED_EXIT); // -sae
+        }
 
         // -s -yqu -yat -yD -D
         if (StringUtils.isNotEmpty(others)) {

dolphinscheduler-task-plugin/dolphinscheduler-task-flink/src/test/java/org/apache/dolphinscheduler/plugin/task/flink/FlinkArgsUtilsTest.java

@@ -68,8 +68,9 @@ public void testRunJarInApplicationMode() throws Exception {
         FlinkParameters flinkParameters = buildTestFlinkParametersWithDeployMode(FlinkDeployMode.APPLICATION);
         List<String> commandLine = FlinkArgsUtils.buildRunCommandLine(buildTestTaskExecutionContext(), flinkParameters);
 
+        // APPLICATION mode should NOT include -sae parameter (detached mode on YARN)
         Assertions.assertEquals(
-                "${FLINK_HOME}/bin/flink run-application -t yarn-application -ys 4 -ynm demo-app-name -yjm 1024m -ytm 1024m -p 4 -sae -c org.example.Main /opt/job.jar",
+                "${FLINK_HOME}/bin/flink run-application -t yarn-application -ys 4 -ynm demo-app-name -yjm 1024m -ytm 1024m -p 4 -c org.example.Main /opt/job.jar",
                 joinStringListWithSpace(commandLine));
     }