[1.3.7-prepare#5453][Improvement][Task] ds flink task support submit a PyFlink job via the CLI (#5847)

issue #5452
pr #5453

dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/Constants.java

@@ -837,7 +837,7 @@ public final class Constants {
     public static final String FLINK_MAIN_CLASS = "-c";
     public static final String FLINK_PARALLELISM = "-p";
     public static final String FLINK_SHUTDOWN_ON_ATTACHED_EXIT = "-sae";
-
+    public static final String FLINK_PYTHON = "-py";
 
     public static final int[] NOT_TERMINATED_STATES = new int[]{
             ExecutionStatus.SUBMITTED_SUCCESS.ordinal(),

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/utils/FlinkArgsUtils.java

@@ -117,9 +117,15 @@ public class FlinkArgsUtils {
 
         ProgramType programType = param.getProgramType();
         String mainClass = param.getMainClass();
-        if (programType != null && programType != ProgramType.PYTHON && StringUtils.isNotEmpty(mainClass)) {
-            args.add(Constants.FLINK_MAIN_CLASS);    //-c
-            args.add(param.getMainClass());          //main class
+
+        if (ProgramType.PYTHON == programType) {
+            // -py
+            args.add(Constants.FLINK_PYTHON);
+        } else if (programType != null && StringUtils.isNotEmpty(mainClass)) {
+            // -c
+            args.add(Constants.FLINK_MAIN_CLASS);
+            // main class
+            args.add(param.getMainClass());
         }
 
         ResourceInfo mainJar = param.getMainJar();