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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
