You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampark.apache.org by li...@apache.org on 2023/09/20 08:52:47 UTC
[incubator-streampark] branch dev updated: [Improve] Add additional use cases for FlinkOperator on Kubernetes (#3152)
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 <77...@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",