[Flink] Fix REMOTE Session JobGraph submission for Flink 1.x/2.x SQL and uploaded JAR jobs

streampark-common/src/main/java/org/apache/streampark/common/util/DeflaterUtils.java

@@ -53,7 +53,12 @@ public static String zipString(String text) {
     }
 
     public static String unzipString(String zipString) {
-        byte[] decode = Base64.getDecoder().decode(zipString);
+        byte[] decode;
+        try {
+            decode = Base64.getDecoder().decode(zipString);
+        } catch (IllegalArgumentException e) {
+            return null;
+        }
         Inflater inflater = new Inflater();
         inflater.setInput(decode);
         byte[] bytes = new byte[256];
@@ -71,4 +76,27 @@ public static String unzipString(String zipString) {
         }
         return outputStream.toString();
     }
+
+    /** Returns plain text, repeatedly decoding when the input is compressed multiple times. */
+    public static String toPlainText(String text) {
+        if (StringUtils.isBlank(text)) {
+            return text;
+        }
+        String current = text;
+        while (true) {
+            String decoded = unzipString(current);
+            if (decoded == null) {
+                return current;
+            }
+            current = decoded;
+        }
+    }
+
+    /** Stores SQL/conf text as a single compressed blob regardless of input encoding. */
+    public static String compressForStorage(String text) {
+        if (StringUtils.isBlank(text)) {
+            return "";
+        }
+        return zipString(toPlainText(text));
+    }
 }

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/assembler/FlinkApplicationAssembler.java

@@ -86,9 +86,11 @@
         FlinkApplication app = new FlinkApplication();
         app.setId(request.getId());
         app.setTeamId(request.getTeamId());
-        app.setRestoreOrTriggerSavepoint(request.getRestoreOrTriggerSavepoint());
+        app.setRestoreOrTriggerSavepoint(
+            request.getRestoreOrTriggerSavepoint() != null ? request.getRestoreOrTriggerSavepoint() : false);
         app.setSavepointPath(request.getSavepointPath());
-        app.setAllowNonRestored(request.getAllowNonRestored());
+        app.setAllowNonRestored(
+            request.getAllowNonRestored() != null ? request.getAllowNonRestored() : false);
         return app;
     }
 
@@ -99,9 +101,10 @@
         FlinkApplication app = new FlinkApplication();
         app.setId(request.getId());
         app.setTeamId(request.getTeamId());
-        app.setRestoreOrTriggerSavepoint(request.getRestoreOrTriggerSavepoint());
-        app.setDrain(request.getDrain());
-        app.setNativeFormat(request.getNativeFormat());
+        app.setRestoreOrTriggerSavepoint(
+            request.getRestoreOrTriggerSavepoint() != null ? request.getRestoreOrTriggerSavepoint() : false);
+        app.setDrain(request.getDrain() != null ? request.getDrain() : false);
+        app.setNativeFormat(request.getNativeFormat() != null ? request.getNativeFormat() : false);
         app.setSavepointPath(request.getSavepointPath());
         return app;
     }

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/FlinkSql.java

@@ -21,6 +21,8 @@
 import org.apache.streampark.console.core.bean.Dependency;
 import org.apache.streampark.console.core.enums.ChangeTypeEnum;
 
+import org.apache.commons.lang3.StringUtils;
+
 import com.baomidou.mybatisplus.annotation.IdType;
 import com.baomidou.mybatisplus.annotation.TableField;
 import com.baomidou.mybatisplus.annotation.TableId;
@@ -85,7 +87,7 @@
     }
 
     public void decode() {
-        this.setSql(DeflaterUtils.unzipString(this.sql));
+        this.setSql(DeflaterUtils.toPlainText(this.sql));
     }
 
     public void setToApplication(FlinkApplication application) {
@@ -105,8 +107,17 @@
     }
 
     public ChangeTypeEnum checkChange(FlinkSql target) {
+        if (target == null) {
+            return ChangeTypeEnum.NONE;
+        }
         // 1) determine if sql statement has changed
-        boolean sqlDifference = !this.getSql().trim().equals(target.getSql().trim());
+        String sourceSql =
+            StringUtils.trimToEmpty(DeflaterUtils.toPlainText(this.getSql()));
+        String targetSql =
+            target.getSql() == null
+                ? sourceSql
+                : StringUtils.trimToEmpty(DeflaterUtils.toPlainText(target.getSql()));
+        boolean sqlDifference = !sourceSql.equals(targetSql);
         // 2) determine if dependency has changed
         Dependency thisDependency = Dependency.toDependency(this.getDependency());
         Dependency targetDependency = Dependency.toDependency(target.getDependency());

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/runner/EnvInitializer.java

@@ -74,6 +74,10 @@ public class EnvInitializer implements ApplicationRunner {
         "^streampark-flink-shims_flink-(1\\.1[7-9]|1\\.2[0-9]|2\\.[0-3])-(.*).jar$",
         Pattern.CASE_INSENSITIVE | Pattern.DOTALL);
 
+    private static final Pattern PATTERN_FLINK_SHIMS_BASE_JAR = Pattern.compile(
+        "^streampark-flink-shims-base(-v2)?-(.*)\\.jar$",
+        Pattern.CASE_INSENSITIVE | Pattern.DOTALL);
+
     @SneakyThrows
     @Override
     public void run(ApplicationArguments args) throws Exception {
@@ -184,6 +188,10 @@ private void uploadShimsJar(Workspace workspace, FsOperator fsOperator) {
         String appShims = workspace.APP_SHIMS();
         fsOperator.delete(appShims);
 
+        File[] shimsBaseJars =
+            WebUtils.getAppLibDir().listFiles(
+                pathname -> pathname.getName().matches(PATTERN_FLINK_SHIMS_BASE_JAR.pattern()));
+
         for (File file : shims) {
             Matcher matcher = PATTERN_FLINK_SHIMS_JAR.matcher(file.getName());
             if (matcher.matches()) {
@@ -192,10 +200,26 @@ private void uploadShimsJar(Workspace workspace, FsOperator fsOperator) {
                 fsOperator.mkdirs(shimsPath);
                 log.info("load shims:{} to {}", file.getName(), shimsPath);
                 fsOperator.upload(file.getAbsolutePath(), shimsPath);
+                if (isFlink2MajorVersion(version) && shimsBaseJars != null) {
+                    uploadShimsBaseJars(fsOperator, shimsBaseJars, shimsPath);
+                }
             }
         }
     }
 
+    private static boolean isFlink2MajorVersion(String majorVersion) {
+        int dot = majorVersion.indexOf('.');
+        String major = dot < 0 ? majorVersion : majorVersion.substring(0, dot);
+        return "2".equals(major);
+    }
+
+    private void uploadShimsBaseJars(FsOperator fsOperator, File[] shimsBaseJars, String shimsPath) {
+        for (File baseJar : shimsBaseJars) {
+            log.info("load shims base:{} to {}", baseJar.getName(), shimsPath);
+            fsOperator.upload(baseJar.getAbsolutePath(), shimsPath);
+        }
+    }
+
     public void checkFlinkEnv(StorageType storageType, FlinkEnv flinkEnv) throws IOException {
         String flinkLocalHome = flinkEnv.getFlinkHome();
         if (StringUtils.isBlank(flinkLocalHome)) {

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/FlinkApplicationActionServiceImpl.java

@@ -261,7 +261,7 @@ public void cancel(FlinkApplication appParam) throws Exception {
         applicationLog.setClusterId(application.getClusterId());
         applicationLog.setUserId(ServiceHelper.getUserId());
 
-        if (appParam.getRestoreOrTriggerSavepoint()) {
+        if (Boolean.TRUE.equals(appParam.getRestoreOrTriggerSavepoint())) {
             FlinkAppHttpWatcher.addSavepoint(application.getId());
             application.setOptionState(OptionStateEnum.SAVEPOINTING.getValue());
         } else {
@@ -280,7 +280,7 @@ public void cancel(FlinkApplication appParam) throws Exception {
 
         // infer savepoint
         String customSavepoint = null;
-        if (appParam.getRestoreOrTriggerSavepoint()) {
+        if (Boolean.TRUE.equals(appParam.getRestoreOrTriggerSavepoint())) {
             customSavepoint = appParam.getSavepointPath();
             if (StringUtils.isBlank(customSavepoint)) {
                 customSavepoint = savepointService.getSavePointPath(appParam);
@@ -315,10 +315,10 @@ public void cancel(FlinkApplication appParam) throws Exception {
                 properties,
                 new JobClientTarget(clusterId, application.getJobId(), namespace),
                 new SavepointCancelOptions(
-                    appParam.getRestoreOrTriggerSavepoint(),
-                    appParam.getDrain(),
+                    Boolean.TRUE.equals(appParam.getRestoreOrTriggerSavepoint()),
+                    Boolean.TRUE.equals(appParam.getDrain()),
                     customSavepoint,
-                    appParam.getNativeFormat()));
+                    Boolean.TRUE.equals(appParam.getNativeFormat())));
 
         final Date triggerTime = new Date();
         CompletableFuture<CancelResponse> cancelFuture =
@@ -344,7 +344,7 @@ public void cancel(FlinkApplication appParam) throws Exception {
                         application.setState(FlinkAppStateEnum.FAILED.getValue());
                         updateById(application);
 
-                        if (appParam.getRestoreOrTriggerSavepoint()) {
+                        if (Boolean.TRUE.equals(appParam.getRestoreOrTriggerSavepoint())) {
                             savepointService.expire(application.getId());
                         }
                         // re-tracking flink job on kubernetes and logging exception
@@ -436,10 +436,8 @@ public void start(FlinkApplication appParam, boolean auto) throws Exception {
         String flinkUserJar = userJarAndAppConf.t1;
         String appConf = userJarAndAppConf.t2;
 
-        BuildResult buildResult = buildPipeline.getBuildResult();
-        if (FlinkDeployMode.YARN_APPLICATION == application.getDeployModeEnum()) {
-            buildResult = new ShadedBuildResponse(null, flinkUserJar, true);
-        }
+        BuildResult buildResult =
+            resolveBuildResultForSubmit(application, buildPipeline.getBuildResult(), flinkUserJar);
 
         // Get the args after placeholder replacement
         String args =
@@ -637,6 +635,41 @@ private void starting(FlinkApplication application) {
         updateById(application);
     }
 
+    private BuildResult resolveBuildResultForSubmit(
+                                                    FlinkApplication application,
+                                                    BuildResult buildResult,
+                                                    String flinkUserJar) {
+        if (FlinkDeployMode.YARN_APPLICATION == application.getDeployModeEnum()) {
+            return new ShadedBuildResponse(null, flinkUserJar, true);
+        }
+        if (!(buildResult instanceof ShadedBuildResponse)) {
+            return buildResult;
+        }
+        ShadedBuildResponse shaded = (ShadedBuildResponse) buildResult;
+        if (StringUtils.isNotBlank(shaded.shadedJarPath())) {
+            return shaded;
+        }
+        String workspace =
+            StringUtils.defaultIfBlank(shaded.workspacePath(), application.getLocalAppHome());
+        String jarPath =
+            workspace
+                + "/streampark-flinkjob_"
+                + application.getJobName().replaceAll("\\s+", "_")
+                + ".jar";
+        File jarFile = new File(jarPath);
+        if (!jarFile.isFile()
+            && application.isUploadResource()
+            && StringUtils.isNotBlank(application.getJar())) {
+            jarFile = new File(workspace, application.getJar());
+        }
+        ApiAlertException.throwIfTrue(
+            !jarFile.isFile(),
+            "[StreamPark] Shaded jar not found at "
+                + jarPath
+                + ", please rebuild the application.");
+        return new ShadedBuildResponse(workspace, jarPath, shaded.pass());
+    }
+
     private Tuple2<String, String> getUserJarAndAppConf(
                                                         FlinkEnv flinkEnv, FlinkApplication application) {
         FlinkDeployMode deployModeEnum = application.getDeployModeEnum();
@@ -835,7 +868,7 @@ private void doAbort(Long id) {
     }
 
     private String getSavepointPath(FlinkApplication appParam) {
-        if (appParam.getRestoreOrTriggerSavepoint()) {
+        if (Boolean.TRUE.equals(appParam.getRestoreOrTriggerSavepoint())) {
             if (StringUtils.isBlank(appParam.getSavepointPath())) {
                 FlinkSavepoint savepoint = savepointService.getLatest(appParam.getId());
                 if (savepoint != null) {

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/FlinkApplicationBuildPipelineServiceImpl.java

@@ -215,7 +215,7 @@ public void onStart(PipelineSnapshot snapshot) {
                             app.getJar(),
                             app.getAppHome(),
                             app.getAppLib(),
-                            app.getDistHome(),
+                            app.isUploadResource() ? null : app.getDistHome(),
                             app.getFsOperator(),
                             appUploads,
                             app.isUploadResource(),

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/FlinkClusterServiceImpl.java

@@ -233,6 +233,9 @@ private void updateFlinkClusterForNonRemoteModes(
     private void updateFlinkClusterForRemoteMode(
                                                  FlinkCluster paramOfCluster, FlinkCluster flinkCluster) {
         flinkCluster.setAddress(paramOfCluster.getAddress());
+        if (paramOfCluster.getVersionId() != null) {
+            flinkCluster.setVersionId(paramOfCluster.getVersionId());
+        }
         flinkCluster.setClusterState(ClusterState.RUNNING.getState());
         flinkCluster.setStartTime(new Date());
         flinkCluster.setEndTime(null);

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/FlinkSqlServiceImpl.java

@@ -72,7 +72,7 @@ public class FlinkSqlServiceImpl extends ServiceImpl<FlinkSqlMapper, FlinkSql>
     public FlinkSql getEffective(Long appId, boolean decode) {
         FlinkSql flinkSql = baseMapper.getEffective(appId);
         if (flinkSql != null && decode) {
-            flinkSql.setSql(DeflaterUtils.unzipString(flinkSql.getSql()));
+            flinkSql.setSql(DeflaterUtils.toPlainText(flinkSql.getSql()));
         }
         return flinkSql;
     }
@@ -90,7 +90,7 @@ public FlinkSql getLatestFlinkSql(Long appId, boolean decode) {
             .map(
                 flinkSql -> {
                     if (decode) {
-                        flinkSql.setSql(DeflaterUtils.unzipString(flinkSql.getSql()));
+                        flinkSql.setSql(DeflaterUtils.toPlainText(flinkSql.getSql()));
                     }
                     return flinkSql;
                 })
@@ -101,8 +101,7 @@ public FlinkSql getLatestFlinkSql(Long appId, boolean decode) {
     public void create(FlinkSql flinkSql) {
         Integer version = this.baseMapper.getLatestVersion(flinkSql.getAppId());
         flinkSql.setVersion(version == null ? 1 : version + 1);
-        String sql = DeflaterUtils.zipString(flinkSql.getSql());
-        flinkSql.setSql(sql);
+        flinkSql.setSql(DeflaterUtils.compressForStorage(flinkSql.getSql()));
         this.save(flinkSql);
         this.setCandidate(CandidateTypeEnum.NEW, flinkSql.getAppId(), flinkSql.getId());
     }

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/util/ApplicationBuildPipelineUtils.java

@@ -155,23 +155,29 @@ public static void prepareJarJobHome(
                                          boolean tempDirFallback) {
         fsOperator.delete(appHome);
         if (!fromUpload) {
+            ApiAlertException.throwIfTrue(
+                StringUtils.isBlank(distHome),
+                "[StreamPark] distHome is required for build-resource jar jobs");
             fsOperator.upload(distHome, appHome);
             return;
         }
 
-        String uploadJar = appUploads.concat("/").concat(jar);
-        File localJar = new File(String.format("%s/%d/%s", Workspace.local().APP_UPLOADS(), teamId, jar));
+        String teamUploads = appUploads.concat("/").concat(String.valueOf(teamId));
+        String uploadJar = teamUploads.concat("/").concat(jar);
+        File localJar = new File(uploadJar);
         if (!localJar.exists()) {
             Resource resource = resourceService.findByResourceName(teamId, jar);
             if (resource != null && StringUtils.isNotBlank(resource.getFilePath())) {
                 localJar = new File(resource.getFilePath());
-                uploadJar = appUploads.concat("/").concat(localJar.getName());
+                uploadJar = teamUploads.concat("/").concat(localJar.getName());
             } else if (tempDirFallback) {
                 localJar = new File(WebUtils.getAppTempDir(), jar);
-                uploadJar = appUploads.concat("/").concat(localJar.getName());
             }
         }
-        checkOrElseUploadJar(fsOperator, localJar, uploadJar, appUploads);
+        if (!localJar.isFile()) {
+            throw new ApiAlertException("Missing jar file: " + jar + ", please upload again");
+        }
+        checkOrElseUploadJar(fsOperator, localJar, uploadJar, teamUploads);
 
         switch (applicationType) {
             case STREAMPARK_FLINK:

streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/assembler/FlinkApplicationAssemblerTest.java

@@ -50,6 +50,18 @@ void shouldConvertCreateRequestToEntity() {
         Assertions.assertEquals(4, app.getDeployMode());
     }
 
+    @Test
+    void shouldDefaultNullStartFlagsToFalse() {
+        FlinkAppStartRequest request = new FlinkAppStartRequest();
+        request.setId(1L);
+        request.setTeamId(100000L);
+
+        FlinkApplication app = FlinkApplicationAssembler.toEntity(request);
+
+        Assertions.assertFalse(app.getRestoreOrTriggerSavepoint());
+        Assertions.assertFalse(app.getAllowNonRestored());
+    }
+
     @Test
     void shouldConvertStartRequestToEntity() {
         FlinkAppStartRequest request = new FlinkAppStartRequest();

streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/java/org/apache/streampark/flink/client/ExitSecurityManager.java

@@ -34,4 +34,9 @@ public void checkExit(int status) {
     public void checkPermission(Permission perm) {
         // no-op
     }
+
+    @Override
+    public void checkPermission(Permission perm, Object context) {
+        // no-op
+    }
 }