Al-assad commented on code in PR #2994:
URL:
https://github.com/apache/incubator-streampark/pull/2994#discussion_r1308674087
##########
streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/KubernetesNativeApplicationClient.scala:
##########
@@ -62,14 +67,17 @@ object KubernetesNativeApplicationClient extends
KubernetesNativeClientTrait {
var clusterDescriptor: KubernetesClusterDescriptor = null
var clusterClient: ClusterClient[String] = null
+ val spec = convertFlinkDeploymentDef(submitRequest, flinkConfig)
+
try {
val (descriptor, clusterSpecification) =
getK8sClusterDescriptorAndSpecification(flinkConfig)
clusterDescriptor = descriptor
- val applicationConfig =
ApplicationConfiguration.fromConfiguration(flinkConfig)
- clusterClient = clusterDescriptor
- .deployApplicationCluster(clusterSpecification, applicationConfig)
- .getClusterClient
+ val applicationConfig =
ApplicationConfiguration.fromConfiguration(flinkConfig)
+// clusterClient = clusterDescriptor
+// .deployApplicationCluster(clusterSpecification, applicationConfig)
+// .getClusterClient
+ FlinkK8sOperator.deployApplicationJob(submitRequest.id, spec)
Review Comment:
FlinkK8sOperator.deployApplicationJob is an IO strucuture, need to invoke
.runIO to run it actually.
##########
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,
Review Comment:
The flink base image should come from buildResult.flinkImageTag, see line 64.
##########
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:
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
