You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampark.apache.org by be...@apache.org on 2022/12/29 09:32:42 UTC
[incubator-streampark] branch javaopts created (now ecfa503bd)
This is an automated email from the ASF dual-hosted git repository.
benjobs pushed a change to branch javaopts
in repository https://gitbox.apache.org/repos/asf/incubator-streampark.git
at ecfa503bd [Feature] jvm options parameter support
This branch includes the following new commits:
new ecfa503bd [Feature] jvm options parameter support
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[incubator-streampark] 01/01: [Feature] jvm options parameter support
Posted by be...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
benjobs pushed a commit to branch javaopts
in repository https://gitbox.apache.org/repos/asf/incubator-streampark.git
commit ecfa503bd4634c450a383e4ca3d1c100c04f9007
Author: benjobs <be...@apache.org>
AuthorDate: Thu Dec 29 17:32:23 2022 +0800
[Feature] jvm options parameter support
---
.../flink/submit/trait/FlinkSubmitTrait.scala | 30 +++++++++++++++++++---
1 file changed, 27 insertions(+), 3 deletions(-)
diff --git a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/trait/FlinkSubmitTrait.scala b/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/trait/FlinkSubmitTrait.scala
index e4e99eda0..227a7eb85 100644
--- a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/trait/FlinkSubmitTrait.scala
+++ b/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/trait/FlinkSubmitTrait.scala
@@ -56,6 +56,7 @@ trait FlinkSubmitTrait extends Logger {
private[submit] lazy val PARAM_KEY_APP_NAME = KEY_APP_NAME("--")
private[submit] lazy val PARAM_KEY_FLINK_PARALLELISM = KEY_FLINK_PARALLELISM("--")
+
@throws[Exception] def submit(submitRequest: SubmitRequest): SubmitResponse = {
logInfo(
s"""
@@ -105,9 +106,33 @@ trait FlinkSubmitTrait extends Logger {
flinkConfig.setBoolean(SavepointConfigOptions.SAVEPOINT_IGNORE_UNCLAIMED_STATE, submitRequest.allowNonRestoredState)
}
+ // set JVMOptions..
+ setJvmOptions(submitRequest, flinkConfig)
+
setConfig(submitRequest, flinkConfig)
doSubmit(submitRequest, flinkConfig)
+
+ }
+
+ def setJvmOptions(submitRequest: SubmitRequest, flinkConfig: Configuration): Unit = {
+ if (MapUtils.isNotEmpty(submitRequest.properties)) {
+ submitRequest.properties.foreach(x => {
+ val k = x._1.trim
+ val v = x._2.toString
+ if (k == CoreOptions.FLINK_JVM_OPTIONS.key()) {
+ flinkConfig.set(CoreOptions.FLINK_JVM_OPTIONS, v)
+ } else if (k == CoreOptions.FLINK_JM_JVM_OPTIONS.key()) {
+ flinkConfig.set(CoreOptions.FLINK_JM_JVM_OPTIONS, v)
+ } else if (k == CoreOptions.FLINK_HS_JVM_OPTIONS.key()) {
+ flinkConfig.set(CoreOptions.FLINK_HS_JVM_OPTIONS, v)
+ } else if (k == CoreOptions.FLINK_TM_JVM_OPTIONS.key()) {
+ flinkConfig.set(CoreOptions.FLINK_TM_JVM_OPTIONS, v)
+ } else if (k == CoreOptions.FLINK_CLI_JVM_OPTIONS.key()) {
+ flinkConfig.set(CoreOptions.FLINK_CLI_JVM_OPTIONS, v)
+ }
+ })
+ }
}
def setConfig(submitRequest: SubmitRequest, flinkConf: Configuration): Unit
@@ -268,9 +293,7 @@ trait FlinkSubmitTrait extends Logger {
// app properties
if (MapUtils.isNotEmpty(submitRequest.properties)) {
submitRequest.properties.foreach(x => {
- if (x._1.startsWith(CoreOptions.FLINK_JVM_OPTIONS.key())) {
- array += s"-D${x._1}=" + "\"" + x._2 + "\""
- } else {
+ if (!x._1.startsWith(CoreOptions.FLINK_JVM_OPTIONS.key())) {
array += s"-D${x._1}=${x._2}"
}
})
@@ -369,6 +392,7 @@ trait FlinkSubmitTrait extends Logger {
processElement(next, false)
}
}
+
processElement(0, false)
argsArray.foreach(x => programArgs += x.trim)
}