Skip to content

Commit f6bd495

Browse files
wolfboyswolfboys
committed
[Improve] Compatibility improvement for parsing Flink config file
1 parent 5acca1d commit f6bd495

13 files changed

Lines changed: 324 additions & 307 deletions

File tree

‎streampark-common/src/main/scala/org/apache/streampark/common/util/FlinkConfigurationUtils.scala‎

Lines changed: 203 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,203 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.streampark.common.util
19+
20+
import org.apache.streampark.common.util.Implicits._
21+
22+
import org.apache.commons.lang3.StringUtils
23+
24+
import javax.annotation.Nonnull
25+
26+
import java.io.File
27+
import java.util.Scanner
28+
import java.util.concurrent.atomic.AtomicInteger
29+
import java.util.regex.Pattern
30+
31+
import scala.collection.mutable
32+
import scala.collection.mutable.ArrayBuffer
33+
34+
object FlinkConfigurationUtils extends Logger {
35+
36+
private[this] val PROPERTY_PATTERN = Pattern.compile("(.*?)=(.*?)")
37+
38+
private[this] val MULTI_PROPERTY_REGEXP = "-D(.*?)\\s*=\\s*[\\\"|'](.*)[\\\"|']"
39+
40+
private[this] val MULTI_PROPERTY_PATTERN = Pattern.compile(MULTI_PROPERTY_REGEXP)
41+
42+
/**
43+
* @param file
44+
* @return
45+
*/
46+
def loadFlinkConf(file: File): JavaMap[String, String] = {
47+
AssertUtils.required(
48+
file != null && file.exists() && file.isFile,
49+
"[StreamPark] loadFlinkConfYaml: file must not be null")
50+
loadFlinkConf(org.apache.commons.io.FileUtils.readFileToString(file))
51+
}
52+
53+
def loadFlinkConf(yaml: String): JavaMap[String, String] = {
54+
AssertUtils.required(yaml != null && yaml.nonEmpty, "[StreamPark] loadFlinkConfYaml: yaml must not be null")
55+
PropertiesUtils.fromYamlText(yaml)
56+
}
57+
58+
def loadLegacyFlinkConf(file: File): JavaMap[String, String] = {
59+
AssertUtils.required(
60+
file != null && file.exists() && file.isFile,
61+
"[StreamPark] loadFlinkConfYaml: file must not be null")
62+
loadLegacyFlinkConf(org.apache.commons.io.FileUtils.readFileToString(file))
63+
}
64+
65+
def loadLegacyFlinkConf(yaml: String): JavaMap[String, String] = {
66+
AssertUtils.required(yaml != null && yaml.nonEmpty, "[StreamPark] loadFlinkConfYaml: yaml must not be null")
67+
val flinkConf = new JavaHashMap[String, String]()
68+
val scanner: Scanner = new Scanner(yaml)
69+
val lineNo: AtomicInteger = new AtomicInteger(0)
70+
while (scanner.hasNextLine) {
71+
val line = scanner.nextLine()
72+
lineNo.incrementAndGet()
73+
// 1. check for comments
74+
// [FLINK-27299] flink parsing parameter bug fixed.
75+
val comments = line.split("^#|\\s+#", 2)
76+
val conf = comments(0).trim
77+
// 2. get key and value
78+
if (conf.nonEmpty) {
79+
val kv = conf.split(": ", 2)
80+
// skip line with no valid key-value pair
81+
if (kv.length == 2) {
82+
val key = kv(0).trim
83+
val value = kv(1).trim
84+
// sanity check
85+
if (key.nonEmpty && value.nonEmpty) {
86+
flinkConf += key -> value
87+
} else {
88+
logWarn(s"Error after splitting key and value in configuration ${lineNo.get()}: $line")
89+
}
90+
} else {
91+
logWarn(s"Error while trying to split key and value in configuration. $lineNo : $line")
92+
}
93+
}
94+
}
95+
flinkConf
96+
}
97+
98+
/** extract flink configuration from application.properties */
99+
@Nonnull def extractDynamicProperties(properties: String): Map[String, String] = {
100+
if (StringUtils.isEmpty(properties)) Map.empty[String, String]
101+
else {
102+
val map = mutable.Map[String, String]()
103+
val simple = properties.replaceAll(MULTI_PROPERTY_REGEXP, "")
104+
simple.split("\\s?-D") match {
105+
case d if Utils.isNotEmpty(d) =>
106+
d.foreach(x => {
107+
if (x.nonEmpty) {
108+
val p = PROPERTY_PATTERN.matcher(x.trim)
109+
if (p.matches) {
110+
map += p.group(1).trim -> p.group(2).trim
111+
}
112+
}
113+
})
114+
case _ =>
115+
}
116+
val matcher = MULTI_PROPERTY_PATTERN.matcher(properties)
117+
while (matcher.find()) {
118+
val opts = matcher.group()
119+
val index = opts.indexOf("=")
120+
val key = opts.substring(2, index).trim
121+
val value =
122+
opts.substring(index + 1).trim.replaceAll("(^[\"|']|[\"|']$)", "")
123+
map += key -> value
124+
}
125+
map.toMap
126+
}
127+
}
128+
129+
@Nonnull def extractArguments(args: String): List[String] = {
130+
val programArgs = new ArrayBuffer[String]()
131+
if (StringUtils.isNotEmpty(args)) {
132+
return extractArguments(args.split("\\s+"))
133+
}
134+
programArgs.toList
135+
}
136+
137+
def extractArguments(array: Array[String]): List[String] = {
138+
val programArgs = new ArrayBuffer[String]()
139+
val iter = array.iterator
140+
while (iter.hasNext) {
141+
val v = iter.next()
142+
val p = v.take(1)
143+
p match {
144+
case "'" | "\"" =>
145+
var value = v
146+
if (!v.endsWith(p)) {
147+
while (!value.endsWith(p) && iter.hasNext) {
148+
value += s" ${iter.next()}"
149+
}
150+
}
151+
programArgs += value.substring(1, value.length - 1)
152+
case _ =>
153+
val regexp = "(.*)='(.*)'$"
154+
if (v.matches(regexp)) {
155+
programArgs += v.replaceAll(regexp, "$1=$2")
156+
} else {
157+
val regexp = "(.*)=\"(.*)\"$"
158+
if (v.matches(regexp)) {
159+
programArgs += v.replaceAll(regexp, "$1=$2")
160+
} else {
161+
programArgs += v
162+
}
163+
}
164+
}
165+
}
166+
programArgs.toList
167+
}
168+
169+
def extractMultipleArguments(array: Array[String]): Map[String, Map[String, String]] = {
170+
val iter = array.iterator
171+
val map = mutable.Map[String, mutable.Map[String, String]]()
172+
while (iter.hasNext) {
173+
val v = iter.next()
174+
v.take(2) match {
175+
case "--" =>
176+
val kv = iter.next()
177+
val regexp = "(.*)=(.*)"
178+
if (kv.matches(regexp)) {
179+
val values = kv.split("=")
180+
val k1 = values(0).trim
181+
val v1 = values(1).replaceAll("^['|\"]|['|\"]$", "")
182+
val k = v.drop(2)
183+
map.get(k) match {
184+
case Some(m) => m += k1 -> v1
185+
case _ => map += k -> mutable.Map(k1 -> v1)
186+
}
187+
}
188+
case _ =>
189+
}
190+
}
191+
map.map(x => x._1 -> x._2.toMap).toMap
192+
}
193+
194+
@Nonnull def extractDynamicPropertiesAsJava(properties: String): JavaMap[String, String] =
195+
new JavaHashMap[String, String](extractDynamicProperties(properties))
196+
197+
@Nonnull def extractMultipleArgumentsAsJava(args: Array[String]): JavaMap[String, JavaMap[String, String]] = {
198+
val map =
199+
extractMultipleArguments(args).map(c => c._1 -> new JavaHashMap[String, String](c._2))
200+
new JavaHashMap[String, JavaMap[String, String]](map)
201+
}
202+
203+
}

0 commit comments

Comments
 (0)