[incubator-streampark] branch dev updated: [Improve] Add additional use cases for FlinkOperator on Kubernetes (#3152)

[linying]

This is an automated email from the ASF dual-hosted git repository.

linying pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-streampark.git


The following commit(s) were added to refs/heads/dev by this push:
     new 3f401a09a [Improve] Add additional use cases for FlinkOperator on 
Kubernetes (#3152)
3f401a09a is described below

commit 3f401a09ac5f5af83c8f1ea24776f4748885de44
Author: caicancai <77189278+caican...@users.noreply.github.com>
AuthorDate: Wed Sep 20 16:52:40 2023 +0800

    [Improve] Add additional use cases for FlinkOperator on Kubernetes (#3152)
    
    * add test
    
    * Change namespace
---
 .../flink/kubernetes/v2/example/UsingOperator.scala | 21 +++++++++++++++++++++
 1 file changed, 21 insertions(+)

diff --git 
a/streampark-flink/streampark-flink-kubernetes-v2/streampark-flink-kubernetes-core/src/test/scala/org/apache/streampark/flink/kubernetes/v2/example/UsingOperator.scala
 
b/streampark-flink/streampark-flink-kubernetes-v2/streampark-flink-kubernetes-core/src/test/scala/org/apache/streampark/flink/kubernetes/v2/example/UsingOperator.scala
index 76245cc71..c00580dad 100644
--- 
a/streampark-flink/streampark-flink-kubernetes-v2/streampark-flink-kubernetes-core/src/test/scala/org/apache/streampark/flink/kubernetes/v2/example/UsingOperator.scala
+++ 
b/streampark-flink/streampark-flink-kubernetes-v2/streampark-flink-kubernetes-core/src/test/scala/org/apache/streampark/flink/kubernetes/v2/example/UsingOperator.scala
@@ -95,6 +95,27 @@ class UsingOperator extends AnyWordSpecLike with 
BeforeAndAfterAll {
     } yield ()
   }
 
+  "Deploy a simple Flink application job with flinkConfiguration to 
Kubernetes" in unsafeRun {
+    val spec = FlinkDeploymentDef(
+      name = "basic-appjob",
+      namespace = "fdev",
+      image = "flink:1.16",
+      flinkVersion = FlinkVersion.V1_16,
+      flinkConfiguration = Map("taskmanager.numberOfTaskSlots" -> "2"),
+      jobManager = JobManagerDef(cpu = 1, memory = "1024m", "2G"),
+      taskManager = TaskManagerDef(cpu = 1, memory = "1024m", "2G"),
+      job = JobDef(
+        jarURI = 
"local:///opt/flink/examples/streaming/StateMachineExample.jar",
+        parallelism = 1
+      )
+    )
+    for {
+      _ <- FlinkK8sOperator.deployApplicationJob(114514, spec)
+      // subscribe job status
+      _ <- 
FlinkK8sObserver.evaluatedJobSnaps.flatSubscribeValues().debugPretty.runDrain
+    } yield ()
+  }
+
   "Deploy an application mode job with additional jar resources such as 
third-party dependencies pr udf" in unsafeRun {
     val spec = FlinkDeploymentDef(
       name = "appjob-with-extra-jar",