[Flink] Fix Flink 2.x Flink SQL on Yarn/K8s Application mode

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

@@ -29,7 +29,7 @@
 import org.apache.streampark.shaded.org.slf4j.spi.LoggerFactoryBinder;
 
 import java.io.ByteArrayInputStream;
-import java.io.File;
+import java.io.InputStream;
 import java.net.URL;
 import java.nio.charset.StandardCharsets;
 
@@ -97,27 +97,26 @@ private ShadedContextInitializer(LoggerContext loggerContext) {
         public void configureByResource(URL url) {
             AssertUtils.notNull(url, "URL argument cannot be null");
             String path = url.getPath();
-            if (path.endsWith("xml")) {
-                JoranConfigurator configurator = new JoranConfigurator();
-                configurator.setContext(loggerContext);
-                try {
-                    String text =
-                        FileUtils.readFile(new File(path))
-                            .replaceAll("org.slf4j", SHADED_PACKAGE + ".org.slf4j")
-                            .replaceAll("ch.qos.logback", SHADED_PACKAGE + ".ch.qos.logback")
-                            .replaceAll("org.apache.log4j", SHADED_PACKAGE + ".org.apache.log4j");
-                    ByteArrayInputStream input =
-                        new ByteArrayInputStream(text.getBytes(StandardCharsets.UTF_8));
-                    configurator.doConfigure(input);
-                } catch (Exception e) {
-                    throw new LogbackException("Failed to configure logger context from " + url, e);
-                }
-            } else {
+            if (path == null || !path.endsWith("xml")) {
                 throw new LogbackException(
                     "Unexpected filename extension of file ["
                         + url
                         + "]. Should be .xml");
             }
+            JoranConfigurator configurator = new JoranConfigurator();
+            configurator.setContext(loggerContext);
+            try (InputStream inputStream = url.openStream()) {
+                String text =
+                    new String(inputStream.readAllBytes(), StandardCharsets.UTF_8)
+                        .replaceAll("org.slf4j", SHADED_PACKAGE + ".org.slf4j")
+                        .replaceAll("ch.qos.logback", SHADED_PACKAGE + ".ch.qos.logback")
+                        .replaceAll("org.apache.log4j", SHADED_PACKAGE + ".org.apache.log4j");
+                ByteArrayInputStream input =
+                    new ByteArrayInputStream(text.getBytes(StandardCharsets.UTF_8));
+                configurator.doConfigure(input);
+            } catch (Exception e) {
+                throw new LogbackException("Failed to configure logger context from " + url, e);
+            }
         }
     }
 }

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

@@ -137,55 +137,55 @@
         FlinkApplicationActionService {
 
     @Qualifier("streamparkDeployExecutor")
     @Autowired
     private Executor executorService;
 
     @Autowired
     private FlinkApplicationBackupService backUpService;
 
     @Autowired
     private FlinkApplicationManageService applicationManageService;
 
     @Autowired
     private FlinkApplicationInfoService applicationInfoService;
 
     @Autowired
     private FlinkApplicationConfigService configService;
 
     @Autowired
     private ApplicationLogService applicationLogService;
 
     @Autowired
     private FlinkEnvService flinkEnvService;
 
     @Autowired
     private FlinkSqlService flinkSqlService;
 
     @Autowired
     private SavepointService savepointService;
 
     @Autowired
     private SettingService settingService;
 
     @Autowired
     private FlinkK8sWatcher k8SFlinkTrackMonitor;
 
     @Autowired
     private FlinkApplicationBuildPipelineService appBuildPipeService;
 
     @Autowired
     private FlinkClusterService flinkClusterService;
 
     @Autowired
     private VariableService variableService;
 
     @Autowired
     private ResourceService resourceService;
 
     @Autowired
     private FlinkClusterWatcher flinkClusterWatcher;
 
     @Autowired
     private FlinkK8sWatcherWrapper k8sWatcherWrapper;
 
     private final Map<Long, CompletableFuture<SubmitResponse>> startFutureMap =
@@ -653,7 +653,7 @@
                 FlinkSql flinkSql = flinkSqlService.getEffective(application.getId(), false);
                 AssertUtils.notNull(flinkSql);
                 // 1) dist_userJar
                 String sqlDistJar = ServiceHelper.getFlinkSqlClientJar(flinkEnv);
                 // 2) appConfig
                 appConf =
                     applicationConfig == null
@@ -661,13 +661,16 @@
                         : String.format("yaml://%s", applicationConfig.getContent());
                 // 3) client
                 if (FlinkDeployMode.YARN_APPLICATION == deployModeEnum) {
-                    String clientPath = Workspace.remote().APP_CLIENT();
-                    flinkUserJar = String.format("%s/%s", clientPath, sqlDistJar);
+                    flinkUserJar =
+                        application.getAppHome()
+                            + "/streampark-flinkjob_"
+                            + application.getJobName().replaceAll("\\s+", "_")
+                            + ".jar";
                 }
                 break;
 
             case PYFLINK:
                 Resource resource =
                     resourceService.findByResourceName(application.getTeamId(), application.getJar());
 
                 ApiAlertException.throwIfNull(
@@ -719,7 +722,7 @@
                         case STREAMPARK_FLINK:
                             flinkUserJar =
                                 String.format(
                                     "%s/%s",
                                     application.getAppLib(),
                                     application.getModule().concat(Constants.JAR_SUFFIX));
                             break;

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

@@ -98,41 +98,41 @@
     implements
         FlinkApplicationBuildPipelineService {
 
     @Autowired
     private FlinkEnvService flinkEnvService;
 
     @Autowired
     private FlinkSqlService flinkSqlService;
 
     @Autowired
     private FlinkApplicationBackupService backUpService;
 
     @Autowired
     private SettingService settingService;
 
     @Autowired
     private MessageService messageService;
 
     @Autowired
     private FlinkApplicationManageService applicationManageService;
 
     @Autowired
     private FlinkApplicationInfoService applicationInfoService;
 
     @Autowired
     private ApplicationLogService applicationLogService;
 
     @Autowired
     private FlinkAppHttpWatcher flinkAppHttpWatcher;
 
     @Autowired
     private FlinkApplicationConfigService applicationConfigService;
 
     @Autowired
     private ResourceService resourceService;
 
     @Qualifier("streamparkBuildPipelineExecutor")
     @Autowired
     private ExecutorService executorService;
 
     private static final Cache<Long, DockerPullSnapshot> DOCKER_PULL_PG_SNAPSHOTS = Caffeine.newBuilder()
@@ -367,8 +367,9 @@
                     yarnProvidedPath = app.getAppHome();
                     localWorkspace = app.getLocalAppHome();
                 }
-                FlinkYarnApplicationBuildRequest yarnAppRequest = buildFlinkYarnApplicationBuildRequest(app, mainClass,
-                    localWorkspace, yarnProvidedPath);
+                FlinkYarnApplicationBuildRequest yarnAppRequest =
+                    buildFlinkYarnApplicationBuildRequest(
+                        app, mainClass, localWorkspace, yarnProvidedPath, flinkEnv);
                 log.info("Submit params to building pipeline : {}", yarnAppRequest);
                 return FlinkYarnApplicationBuildPipeline.of(yarnAppRequest);
             case YARN_PER_JOB:
@@ -400,13 +401,21 @@
                                                                                    @Nonnull FlinkApplication app,
                                                                                    String mainClass,
                                                                                    String localWorkspace,
-                                                                                   String yarnProvidedPath) {
+                                                                                   String yarnProvidedPath,
+                                                                                   FlinkEnv flinkEnv) {
+        String sqlDistJar = ServiceHelper.getFlinkSqlClientJar(flinkEnv);
+        String localSqlClientJar =
+            Workspace.local().APP_CLIENT().concat("/").concat(sqlDistJar);
         return new FlinkYarnApplicationBuildRequest(
             app.getJobName(),
             mainClass,
+            app.getLocalAppHome(),
             localWorkspace,
             yarnProvidedPath,
+            localSqlClientJar,
             app.getJobTypeEnum(),
+            app.getDeployModeEnum(),
+            flinkEnv.getFlinkVersion(),
             getMergedDependencyInfo(app));
     }
 

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

@@ -78,19 +78,19 @@
         "Queue label '%s' isn't available in database, please add it first.";
 
     @Qualifier("streamparkClusterExecutor")
     @Autowired
     private ExecutorService executorService;
 
     @Autowired
     private FlinkEnvService flinkEnvService;
 
     @Autowired
     private FlinkApplicationInfoService applicationInfoService;
 
     @Autowired
     private YarnQueueService yarnQueueService;
 
     @Autowired
     private FlinkClusterWatcher flinkClusterWatcher;
 
     @Override
@@ -417,7 +417,7 @@
             flinkCluster.getId(),
             getKubernetesDeployDesc(flinkCluster, "shutdown"));
         Future<ShutDownResponse> future = executorService.submit(() -> FlinkClient.shutdown(stopRequest));
-        return future.get(60, TimeUnit.SECONDS);
+        return future.get(300, TimeUnit.SECONDS);
     }
 
     private DeployResponse deployInternal(FlinkCluster flinkCluster) throws InterruptedException, ExecutionException, TimeoutException {
@@ -430,7 +430,7 @@
             getKubernetesDeployDesc(flinkCluster, "start"));
         log.info("Deploy cluster request {}", deployRequest);
         Future<DeployResponse> future = executorService.submit(() -> FlinkClient.deploy(deployRequest));
-        return future.get(60, TimeUnit.SECONDS);
+        return future.get(300, TimeUnit.SECONDS);
     }
 
     private void checkActiveIfNeeded(FlinkCluster flinkCluster) {

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/watcher/FlinkClusterWatcher.java

@@ -62,19 +62,19 @@
 @Component
 public class FlinkClusterWatcher {
 
     @Autowired
     private FlinkClusterService flinkClusterService;
 
     @Autowired
     private AlertService alertService;
 
     @Autowired
     private FlinkApplicationInfoService applicationInfoService;
 
     private static final Timeout HTTP_TIMEOUT = Timeout.ofSeconds(5L);
 
     @Qualifier("flinkClusterWatchingExecutor")
     @Autowired
     private Executor executorService;
 
     private Long lastWatchTime = 0L;
@@ -196,6 +196,7 @@
     private ClusterState httpClusterState(FlinkCluster flinkCluster) {
         switch (flinkCluster.getFlinkDeployModeEnum()) {
             case REMOTE:
+            case KUBERNETES_NATIVE_SESSION:
                 return httpRemoteClusterState(flinkCluster);
             case YARN_SESSION:
                 return httpYarnSessionClusterState(flinkCluster);

streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/java/org/apache/streampark/flink/client/impl/KubernetesNativeSessionClient.java

@@ -119,9 +119,9 @@ public SubmitResponse jobGraphSubmit(
                         flinkConfig,
                         jarFile,
                         () -> getK8sClusterDescriptor(flinkConfig)
-                            .retrieve(flinkConfig.getString(KubernetesConfigOptions.CLUSTER_ID))
+                            .retrieve(flinkConfig.get(KubernetesConfigOptions.CLUSTER_ID))
                             .getClusterClient(),
-                        () -> flinkConfig.getString(KubernetesConfigOptions.CLUSTER_ID));
+                        () -> flinkConfig.get(KubernetesConfigOptions.CLUSTER_ID));
                 logInfo(
                     "[flink-submit] flink job has been submitted. "
                         + flinkConfIdentifierInfo(flinkConfig)

streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/java/org/apache/streampark/flink/client/impl/YarnApplicationClient.java

@@ -69,8 +69,10 @@ public void setConfig(SubmitRequest submitRequest, Configuration flinkConfig) {
         providedLibs.add(hdfsWorkspace.appJars());
 
         if (submitRequest.jobType() == FlinkJobType.FLINK_SQL) {
-            providedLibs.add(
-                WORKSPACE.APP_SHIMS() + "/flink-" + submitRequest.flinkVersion().majorVersion());
+            // Flink SQL yarn-application builds a fat jar that already shades version-specific
+            // shims. Adding the HDFS shims directory here makes parent-first classloading pick
+            // shims-base ahead of shims-base-v2 and breaks Flink 2.x (NoSuchMethodError on
+            // FlinkConfiguration).
             String jobLib = WORKSPACE.APP_WORKSPACE() + "/" + submitRequest.id() + "/lib";
             try {
                 if (HdfsUtils.exists(jobLib)) {

streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/java/org/apache/streampark/flink/client/impl/YarnSessionClient.java

@@ -34,11 +34,9 @@
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.client.deployment.ClusterSpecification;
 import org.apache.flink.client.program.ClusterClient;
-import org.apache.flink.client.program.PackagedProgram;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.DeploymentOptions;
 import org.apache.flink.configuration.DeploymentOptionsInternal;
-import org.apache.flink.runtime.jobgraph.JobGraph;
 import org.apache.flink.util.FlinkException;
 import org.apache.flink.yarn.YarnClusterDescriptor;
 import org.apache.flink.yarn.configuration.YarnConfigOptions;
@@ -90,36 +88,17 @@
 
     @Override
     public SubmitResponse doSubmit(SubmitRequest submitRequest, Configuration flinkConfig) throws FlinkException {
-        return callAsFlinkException(
-            () -> {
-                Tuple2<ApplicationId, YarnClusterDescriptor> yarnClusterDescriptor =
-                    getYarnClusterDescriptor(flinkConfig);
-                ApplicationId yarnClusterId = yarnClusterDescriptor._1();
-                YarnClusterDescriptor clusterDescriptor = yarnClusterDescriptor._2();
-
-                Tuple2<PackagedProgram, JobGraph> programJobGraph =
-                    getJobGraph(flinkConfig, submitRequest, submitRequest.userJarFile());
-                PackagedProgram packageProgram = programJobGraph._1();
-                JobGraph jobGraph = programJobGraph._2();
-
-                ClusterClient<ApplicationId> client =
-                    clusterDescriptor.retrieve(yarnClusterId).getClusterClient();
-                String jobId = client.submitJob(jobGraph).get().toString();
-                String jobManagerUrl = client.getWebInterfaceURL();
-
-                logInfo(
-                    String.format(
-                        "%n-------------------------<<applicationId>>------------------------%n"
-                            + "Flink Job Started: jobId: %s , applicationId: %s%n"
-                            + "__________________________________________________________________%n",
-                        jobId, yarnClusterId));
-
-                SubmitResponse resp =
-                    new SubmitResponse(
-                        yarnClusterId.toString(), flinkConfig.toMap(), jobId, jobManagerUrl);
-                closeSubmit(submitRequest, packageProgram, client, clusterDescriptor);
-                return resp;
-            });
+        Tuple2<ApplicationId, YarnClusterDescriptor> yarnClusterDescriptor =
+            getYarnClusterDescriptor(flinkConfig);
+        ApplicationId yarnClusterId = yarnClusterDescriptor._1();
+        YarnClusterDescriptor clusterDescriptor = yarnClusterDescriptor._2();
+        return submitJobGraphToCluster(
+            submitRequest,
+            flinkConfig,
+            submitRequest.userJarFile(),
+            () -> clusterDescriptor.retrieve(yarnClusterId).getClusterClient(),
+            () -> yarnClusterId.toString(),
+            clusterDescriptor);
     }
 
     @Override
@@ -196,7 +175,7 @@
 
     private DeployResponse tryReuseExistingYarnSession(
                                                        DeployRequest deployRequest,
                                                        YarnClusterDescriptor clusterDescriptor) throws Exception {
         try {
             FinalApplicationStatus applicationStatus =
                 clusterDescriptor
@@ -224,7 +203,7 @@
         }
     }
 
     public ShutDownResponse shutdown(ShutDownRequest shutDownRequest) throws Exception {
         YarnClusterDescriptor clusterDescriptor = null;
         ClusterClient<ApplicationId> client = null;
         try {

streampark-flink/streampark-flink-packer/src/main/java/org/apache/streampark/flink/packer/docker/FlinkDockerfileTemplate.java

@@ -62,7 +62,7 @@ public Set<String> flinkExtraLibPaths() {
     public String offerDockerfileContent() {
         return "FROM " + flinkBaseImage + "\n"
             + "RUN mkdir -p " + FLINK_HOME + "/usrlib\n"
-            + "COPY " + extraLibName() + " " + FLINK_HOME + "/lib/\n"
+            + "COPY " + extraLibName() + "/. " + FLINK_HOME + "/lib/\n"
             + "COPY " + mainJarName() + " " + FLINK_HOME + "/usrlib/" + mainJarName() + "\n";
     }
 }

streampark-flink/streampark-flink-packer/src/main/java/org/apache/streampark/flink/packer/docker/FlinkHadoopDockerfileTemplate.java

@@ -106,7 +106,7 @@ public String offerDockerfileContent() {
         dockerfile
             .append("COPY ")
             .append(extraLibName())
-            .append(" ")
+            .append("/. ")
             .append(FLINK_HOME)
             .append("/lib/\n")
             .append("COPY ")

streampark-flink/streampark-flink-packer/src/main/java/org/apache/streampark/flink/packer/maven/MavenTool.java

@@ -57,6 +57,7 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -145,7 +146,7 @@
                 "[StreamPark] streampark-packer: failed to delete existing uber jar: "
                     + outFatJarPath);
         }
-        Set<File> jarSet = new HashSet<>();
+        Set<File> jarSet = new LinkedHashSet<>();
         for (String lib : jarLibs) {
             File libFile = new File(lib);
             if (!libFile.exists()) {
@@ -156,11 +157,14 @@
             } else if (libFile.isDirectory()) {
                 File[] files = libFile.listFiles();
                 if (files != null) {
+                    List<File> dirJars = new ArrayList<>();
                     for (File f : files) {
                         if (isJarFile(f)) {
-                            jarSet.add(f);
+                            dirJars.add(f);
                         }
                     }
+                    dirJars.sort(ShimJarMergeOrder.COMPARATOR);
+                    jarSet.addAll(dirJars);
                 }
             }
         }
@@ -219,7 +223,7 @@
      * necessary. Only compile scope dependencies will be resolved.
      */
     @Nonnull
     public static List<File> resolveArtifacts(@Nullable Set<Artifact> mavenArtifacts) throws Exception {
         if (mavenArtifacts == null) {
             return Collections.emptyList();
         }

streampark-flink/streampark-flink-packer/src/main/java/org/apache/streampark/flink/packer/maven/ShimJarMergeOrder.java

@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.streampark.flink.packer.maven;
+
+import java.io.File;
+import java.util.Comparator;
+
+/** Controls duplicate class resolution when shading StreamPark shims into a fat jar. */
+final class ShimJarMergeOrder {
+
+    private static final String SHIMS_BASE_PREFIX = "streampark-flink-shims-base-";
+    private static final String SHIMS_BASE_V2_PREFIX = "streampark-flink-shims-base-v2-";
+
+    static final Comparator<File> COMPARATOR = Comparator.comparingInt(ShimJarMergeOrder::mergeOrder);
+
+    private ShimJarMergeOrder() {
+    }
+
+    private static int mergeOrder(File jar) {
+        String name = jar.getName();
+        if (name.startsWith(SHIMS_BASE_V2_PREFIX)) {
+            return 0;
+        }
+        if (name.startsWith(SHIMS_BASE_PREFIX)) {
+            return 1;
+        }
+        return 2;
+    }
+}

streampark-flink/streampark-flink-packer/src/main/java/org/apache/streampark/flink/packer/pipeline/FlinkSqlDependencySupport.java

@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.streampark.flink.packer.pipeline;
+
+import org.apache.streampark.flink.packer.maven.Artifact;
+import org.apache.streampark.flink.packer.maven.DependencyInfo;
+import org.apache.streampark.flink.packer.maven.MavenTool;
+
+import java.io.File;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
+/** Shared dependency helpers for Flink SQL build pipelines. */
+public final class FlinkSqlDependencySupport {
+
+    private static final Artifact SNAKEYAML = new Artifact("org.yaml", "snakeyaml", "2.0");
+
+    private FlinkSqlDependencySupport() {
+    }
+
+    public static DependencyInfo withSnakeyaml(DependencyInfo dependencyInfo) {
+        Set<String> extJarLibs = new HashSet<>(dependencyInfo.extJarLibs());
+        addSnakeyaml(extJarLibs);
+        return new DependencyInfo(dependencyInfo.mavenArts(), extJarLibs);
+    }
+
+    public static void addSnakeyaml(Set<String> extJarLibs) {
+        String appHome = System.getProperty("app.home", "/streampark");
+        File snakeyaml = new File(appHome, "lib/snakeyaml-2.0.jar");
+        if (snakeyaml.isFile()) {
+            extJarLibs.add(snakeyaml.getAbsolutePath());
+            return;
+        }
+        try {
+            MavenTool.resolveArtifacts(Collections.singleton(SNAKEYAML))
+                .stream()
+                .map(File::getAbsolutePath)
+                .forEach(extJarLibs::add);
+        } catch (Exception e) {
+            throw new IllegalStateException("Failed to resolve snakeyaml for Flink SQL application build", e);
+        }
+    }
+}