You are viewing a plain text version of this content. The canonical link for it is here.

Posted to issues@streampark.apache.org by "caicancai (via GitHub)" <gi...@apache.org> on 2023/08/29 04:55:29 UTC

[GitHub] [incubator-streampark] caicancai opened a new pull request, #2994: Refactor the lifecycle control of flink application mode -dosubmit

caicancai opened a new pull request, #2994:
URL: https://github.com/apache/incubator-streampark/pull/2994

   <!--
   Thank you for contributing to StreamPark! Please make sure that your code changes
   are covered with tests. And in case of new features or big changes
   remember to adjust the documentation.
   
   ## Contribution Checklist
   
     - If this is your first time, please read our contributor guidelines: [Submit Code](https://streampark.apache.org/community/submit_guide/submit_code).
   
     - Make sure that the pull request corresponds to a [GITHUB issue](https://github.com/apache/incubator-streampark/issues).
   
     - Name the pull request in the form "[Feature] Title of the pull request", where *Feature* can be replaced by `Hotfix`, `Bug`, etc.
   
     - Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
   
     - If the PR is unfinished, add `[WIP]` in your PR title, e.g., `[WIP][Feature] Title of the pull request`.
   
   -->
   
   ## What changes were proposed in this pull request
   
   Issue Number: close #xxx <!-- REMOVE this line if no issue to close -->
   
   <!--(For example: This pull request proposed to add checkstyle plugin).-->
   
   ## Brief change log
   
   <!--*(for example:)*
   - *Add maven-checkstyle-plugin to root pom.xml*
   -->
   
   ## Verifying this change
   
   <!--*(Please pick either of the following options)*-->
   
   This change is a trivial rework / code cleanup without any test coverage.
   
   *(or)*
   
   This change is already covered by existing tests, such as *(please describe tests)*.
   
   *(or)*
   
   This change added tests and can be verified as follows:
   
   <!--*(example:)*
   - *Added integration tests for end-to-end.*
   - *Added *Test to verify the change.*
   - *Manually verified the change by testing locally.* -->
   
   ## Does this pull request potentially affect one of the following parts
    - Dependencies (does it add or upgrade a dependency): (yes / no)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [incubator-streampark] Al-assad commented on a diff in pull request #2994: Refactor the lifecycle control of flink application mode -dosubmit

Posted by "Al-assad (via GitHub)" <gi...@apache.org>.

Al-assad commented on code in PR #2994:
URL: https://github.com/apache/incubator-streampark/pull/2994#discussion_r1315539896


##########
streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/KubernetesNativeApplicationClient.scala:
##########
@@ -103,4 +110,39 @@ object KubernetesNativeApplicationClient extends KubernetesNativeClientTrait {
     flinkConf.safeSet(DeploymentOptions.TARGET, ExecutionMode.KUBERNETES_NATIVE_APPLICATION.getName)
     super.doTriggerSavepoint(request, flinkConf)
   }
+
+  private[this] def convertFlinkDeploymentDef(
+      submitRequest: SubmitRequest,
+      flinkConfig: Configuration): FlinkDeploymentDef = {
+    val spec = FlinkDeploymentDef(
+      name = submitRequest.appName,
+      namespace = submitRequest.k8sSubmitParam.kubernetesNamespace,
+      image = KubernetesConfigOptions.CONTAINER_IMAGE.defaultValue(),

Review Comment:
   The flink base image should come from buildResult.flinkImageTag 



##########
streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/KubernetesNativeApplicationClient.scala:
##########
@@ -59,17 +64,19 @@ object KubernetesNativeApplicationClient extends KubernetesNativeClientTrait {
     flinkConfig.safeSet(KubernetesConfigOptions.CONTAINER_IMAGE, buildResult.flinkImageTag)

Review Comment:
   Do not directly set the image tag into the Flink configuration in the operator API.



##########
streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/KubernetesNativeApplicationClient.scala:
##########
@@ -59,17 +64,19 @@ object KubernetesNativeApplicationClient extends KubernetesNativeClientTrait {
     flinkConfig.safeSet(KubernetesConfigOptions.CONTAINER_IMAGE, buildResult.flinkImageTag)
 
     // retrieve k8s cluster and submit flink job on application mode
-    var clusterDescriptor: KubernetesClusterDescriptor = null
     var clusterClient: ClusterClient[String] = null

Review Comment:
   ClusterClient is unnecessary in flink-k8s v2.



##########
streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/KubernetesNativeApplicationClient.scala:
##########
@@ -84,7 +91,7 @@ object KubernetesNativeApplicationClient extends KubernetesNativeClientTrait {
         logError(s"submit flink job fail in ${submitRequest.executionMode} mode")
         throw e
     } finally {
-      Utils.close(clusterDescriptor, clusterClient)
+      Utils.close(clusterClient)

Review Comment:
   It will cause a NullPointException



##########
streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/KubernetesNativeApplicationClient.scala:
##########
@@ -103,4 +110,39 @@ object KubernetesNativeApplicationClient extends KubernetesNativeClientTrait {
     flinkConf.safeSet(DeploymentOptions.TARGET, ExecutionMode.KUBERNETES_NATIVE_APPLICATION.getName)

Review Comment:
   Do not directly set the TARGET  into the Flink configuration in the operator API.



##########
streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/KubernetesNativeApplicationClient.scala:
##########
@@ -103,4 +110,39 @@ object KubernetesNativeApplicationClient extends KubernetesNativeClientTrait {
     flinkConf.safeSet(DeploymentOptions.TARGET, ExecutionMode.KUBERNETES_NATIVE_APPLICATION.getName)
     super.doTriggerSavepoint(request, flinkConf)
   }
+
+  private[this] def convertFlinkDeploymentDef(
+      submitRequest: SubmitRequest,
+      flinkConfig: Configuration): FlinkDeploymentDef = {
+    val spec = FlinkDeploymentDef(
+      name = submitRequest.appName,
+      namespace = submitRequest.k8sSubmitParam.kubernetesNamespace,
+      image = KubernetesConfigOptions.CONTAINER_IMAGE.defaultValue(),
+      flinkVersion = Option(submitRequest.flinkVersion.majorVersion)
+        .map(_.replace(".", "_"))
+        .map("V" + _)
+        .flatMap(v => FlinkVersion.values().find(_.name() == v)) match {
+        case Some(version) => version
+        case None => throw new IllegalArgumentException("Flink version not found")
+      },
+      jobManager = JobManagerDef(
+        cpu = 1,
+        memory = flinkConfig.get(JobManagerOptions.TOTAL_PROCESS_MEMORY).toString),
+      taskManager = TaskManagerDef(
+        cpu = 1,
+        memory = flinkConfig.get(TaskManagerOptions.TOTAL_PROCESS_MEMORY).toString),
+      job = Some(
+        JobDef(
+          jarURI =
+            submitRequest.buildResult.asInstanceOf[DockerImageBuildResponse].dockerInnerMainJarPath,
+          parallelism = 1
+        )),
+      extJarPaths = submitRequest.userJarFile match {
+        case null => Array.empty[String]
+        case file => Array(file.getAbsolutePath)
+      }
+    )
+    spec

Review Comment:
   Lack of settings for FlinkDeploymentDef.flinkConfiguration, these additional flink user parameters may comes from SubmitRequest.extraParameter.



##########
streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/KubernetesNativeApplicationClient.scala:
##########
@@ -103,4 +110,39 @@ object KubernetesNativeApplicationClient extends KubernetesNativeClientTrait {
     flinkConf.safeSet(DeploymentOptions.TARGET, ExecutionMode.KUBERNETES_NATIVE_APPLICATION.getName)
     super.doTriggerSavepoint(request, flinkConf)
   }
+
+  private[this] def convertFlinkDeploymentDef(
+      submitRequest: SubmitRequest,
+      flinkConfig: Configuration): FlinkDeploymentDef = {
+    val spec = FlinkDeploymentDef(
+      name = submitRequest.appName,
+      namespace = submitRequest.k8sSubmitParam.kubernetesNamespace,
+      image = KubernetesConfigOptions.CONTAINER_IMAGE.defaultValue(),
+      flinkVersion = Option(submitRequest.flinkVersion.majorVersion)
+        .map(_.replace(".", "_"))
+        .map("V" + _)
+        .flatMap(v => FlinkVersion.values().find(_.name() == v)) match {
+        case Some(version) => version
+        case None => throw new IllegalArgumentException("Flink version not found")
+      },
+      jobManager = JobManagerDef(
+        cpu = 1,
+        memory = flinkConfig.get(JobManagerOptions.TOTAL_PROCESS_MEMORY).toString),
+      taskManager = TaskManagerDef(
+        cpu = 1,
+        memory = flinkConfig.get(TaskManagerOptions.TOTAL_PROCESS_MEMORY).toString),
+      job = Some(
+        JobDef(
+          jarURI =
+            submitRequest.buildResult.asInstanceOf[DockerImageBuildResponse].dockerInnerMainJarPath,

Review Comment:
   Missing settings for entryclass and args for JobDef, especially in streampark sql mode. What's more, Don't ignore the savepoint and restoreMode parameter in SubmitRequest.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [incubator-streampark] Al-assad commented on a diff in pull request #2994: Refactor the lifecycle control of flink application mode -dosubmit

Posted by "Al-assad (via GitHub)" <gi...@apache.org>.

Al-assad commented on code in PR #2994:
URL: https://github.com/apache/incubator-streampark/pull/2994#discussion_r1308703813


##########
streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/KubernetesNativeApplicationClient.scala:
##########
@@ -103,4 +111,40 @@ object KubernetesNativeApplicationClient extends KubernetesNativeClientTrait {
     flinkConf.safeSet(DeploymentOptions.TARGET, ExecutionMode.KUBERNETES_NATIVE_APPLICATION.getName)
     super.doTriggerSavepoint(request, flinkConf)
   }
+
+  private[this] def convertFlinkDeploymentDef(
+      submitRequest: SubmitRequest,
+      flinkConfig: Configuration): FlinkDeploymentDef = {
+    val spec = FlinkDeploymentDef(
+      name = submitRequest.appName,
+      namespace = submitRequest.k8sSubmitParam.kubernetesNamespace,
+      image = "flink:" + submitRequest.flinkVersion.majorVersion,
+      flinkVersion = converFlinkVersion(submitRequest.flinkVersion.majorVersion).getOrElse(null),
+      jobManager = JobManagerDef(
+        cpu = 1,
+        memory = flinkConfig.get(JobManagerOptions.TOTAL_PROCESS_MEMORY).toString),
+      taskManager = TaskManagerDef(
+        cpu = 1,
+        memory = flinkConfig.get(TaskManagerOptions.TOTAL_PROCESS_MEMORY).toString),
+      job = Some(
+        JobDef(
+          jarURI =
+            submitRequest.buildResult.asInstanceOf[DockerImageBuildResponse].dockerInnerMainJarPath,
+          parallelism = submitRequest.properties.get("parallelism.default").asInstanceOf[Int]
+        )),
+      extJarPaths = Array(submitRequest.userJarFile.getAbsolutePath)
+    )
+    spec
+  }
+
+  private[this] def converFlinkVersion(version: String): Option[FlinkVersion] = {

Review Comment:
   When adding support for a new version of Flink in the CRD, this part of the code needs to be updated. There is a better way here:
   ```scala
   Option(ubmitRequest.flinkVersion.majorVersion)
       .map(_.replace(".", "_"))
       .map("V" + _ )
       .flatMap(v => Test.values().find(_.name() == v))
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [incubator-streampark] Al-assad commented on a diff in pull request #2994: Refactor the lifecycle control of flink application mode -dosubmit

Posted by "Al-assad (via GitHub)" <gi...@apache.org>.

Al-assad commented on code in PR #2994:
URL: https://github.com/apache/incubator-streampark/pull/2994#discussion_r1312795662


##########
streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/KubernetesNativeApplicationClient.scala:
##########
@@ -62,14 +67,16 @@ object KubernetesNativeApplicationClient extends KubernetesNativeClientTrait {
     var clusterDescriptor: KubernetesClusterDescriptor = null

Review Comment:
   remove this code



##########
streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/KubernetesNativeApplicationClient.scala:
##########
@@ -103,4 +110,37 @@ object KubernetesNativeApplicationClient extends KubernetesNativeClientTrait {
     flinkConf.safeSet(DeploymentOptions.TARGET, ExecutionMode.KUBERNETES_NATIVE_APPLICATION.getName)
     super.doTriggerSavepoint(request, flinkConf)
   }
+
+  private[this] def convertFlinkDeploymentDef(
+      submitRequest: SubmitRequest,
+      flinkConfig: Configuration): FlinkDeploymentDef = {
+    val spec = FlinkDeploymentDef(
+      name = submitRequest.appName,
+      namespace = submitRequest.k8sSubmitParam.kubernetesNamespace,
+      image = KubernetesConfigOptions.CONTAINER_IMAGE.defaultValue(),
+      flinkVersion = Option(submitRequest.flinkVersion.majorVersion)
+        .map(_.replace(".", "_"))
+        .map("V" + _)
+        .flatMap(v => FlinkVersion.values().find(_.name() == v))
+        .getOrElse(null),

Review Comment:
   flinkVersion should not be null in the k8s operator, when it get a null value, you should throw an exception to interrupt the doSubmit process.



##########
streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/KubernetesNativeApplicationClient.scala:
##########
@@ -62,14 +67,16 @@ object KubernetesNativeApplicationClient extends KubernetesNativeClientTrait {
     var clusterDescriptor: KubernetesClusterDescriptor = null
     var clusterClient: ClusterClient[String] = null
 
+    val spec: FlinkDeploymentDef = convertFlinkDeploymentDef(submitRequest, flinkConfig)
     try {
-      val (descriptor, clusterSpecification) = getK8sClusterDescriptorAndSpecification(flinkConfig)
-      clusterDescriptor = descriptor
-      val applicationConfig = ApplicationConfiguration.fromConfiguration(flinkConfig)
-      clusterClient = clusterDescriptor
-        .deployApplicationCluster(clusterSpecification, applicationConfig)
-        .getClusterClient
-
+//      val (descriptor, clusterSpecification) = getK8sClusterDescriptorAndSpecification(flinkConfig)
+//      clusterDescriptor = descriptor
+//
+//      val applicationConfig = ApplicationConfiguration.fromConfiguration(flinkConfig)
+//      clusterClient = clusterDescriptor
+//        .deployApplicationCluster(clusterSpecification, applicationConfig)
+//        .getClusterClient
+      FlinkK8sOperator.deployApplicationJob(submitRequest.id, spec).runIO
       val clusterId = clusterClient.getClusterId

Review Comment:
   This code causes a potential NullPointException



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org