研究內容
flink客戶端提交命令為 flink run ...., 如果客戶端的main 需要讀取系統屬性(System properties)，讀取系統屬性變量的位置有兩種：
- 從作業的main方法中讀取。
- 從作業的算子中讀取。
測試環境
Flink -m yarn-cluster
測試方法
對於JVM參數指定的系統變量
-Dkafka.start_from_timestamp=1648828800000
指定方式
FLINK_ENV_JAVA_OPTS="-Dkafka.start_from_timestamp=1648828800009" bin/flink run ...
讀取環境變量
String property = System.getProperty("kafka.start_from_timestamp");
if (property == null) {
//-Dkafka.start_from_timestamp=1648828800000
System.err.println("-Dkafka.start_from_timestamp Not found");
System.err.println("This are Properties Found in this JVM:");
System.err.println(System.getProperties().stringPropertyNames());
} else {
System.err.println("-Dkafka.start_from_timestamp is" + property);
}
-
讀取位置1（在算子外部）提交日志中的輸出：
-Dkafka.start_from_timestamp is1648828800009 -
讀取位置2（在算子內部）TaskManager 輸出結果：
-Dkafka.start_from_timestamp Not found
This are Properties Found in this JVM:
[zookeeper.sasl.client, java.runtime.name, sun.boot.library.path, java.vm.version, java.vm.vendor, java.vendor.url, path.separator, java.vm.name, file.encoding.pkg, user.country, sun.java.launcher, sun.os.patch.level, java.vm.specification.name, user.dir, java.runtime.version, java.awt.graphicsenv, java.endorsed.dirs, os.arch, java.io.tmpdir, line.separator, java.vm.specification.vendor, os.name, log4j.configuration, sun.jnu.encoding, java.library.path, sun.nio.ch.bugLevel, java.specification.name, java.class.version, sun.management.compiler, os.version, user.home, user.timezone, java.awt.printerjob, file.encoding, java.specification.version, log4j.configurationFile, user.name, java.class.path, log.file, java.vm.specification.version, sun.arch.data.model, java.home, sun.java.command, java.specification.vendor, user.language, awt.toolkit, java.vm.info, java.version, java.ext.dirs, sun.boot.class.path, java.vendor, java.security.auth.login.config, file.separator, java.vendor.url.bug, sun.cpu.endian, sun.io.unicode.encoding, sun.cpu.isalist]
測試項目
1. flink-conf.yaml 中指定 env.java.opts
2. FLINK_ENV_JAVA_OPTS 指定 -Dkey=value這樣的 System Properties
3. 在 flink run -m yarn-cluster ... -yD env.java.opts="自定義參數" 中這樣指定
注：env.java.opts.client, env.java.opts.taskmanager 的測試方法均類似
本文僅以第二種指定方式為例。其他方式不做贅述。
結論
- 對於 FLINK_ENV_JAVA_OPTS 系統環境變量設置的自定義系統變量，僅在客戶端提交作業過程中可以訪問。
- 對於 flink-conf.yaml 中的 env.java.opts 經過驗證，可以在客戶端以及TaskManager中訪問。
- 對於 flink run -m yarn-cluster ... -yD env.java.opts="自定義參數"，僅在算子也就是在TaskManager中可以訪問，客戶端中無法訪問。
感嘆
JVM參數，系統屬性，只不過是Java的一個option而已。
回到起點
java [options] classname [args]
-Dproperty=value
Sets a system property value. The property variable is a string with no spaces that represents the name of the property. The value variable is a string that represents the value of the property. If value is a string with spaces, then enclose it in quotation marks (for example -Dfoo="foo bar").
也就是說，必須在 緊跟 java 才能生效，舉例：
無效
java -jar -Dxxx=yyy example.jar
有效
java -Dxxx=yyy -jar example.jar
對於bin/flink：
exec "${JAVA_RUN}" $JVM_ARGS $FLINK_ENV_JAVA_OPTS "${log_setting[@]}" -classpath "`manglePathList "$CC_CLASSPATH:$INTERNAL_HADOOP_CLASSPATHS"`" org.apache.flink.client.cli.CliFrontend "$@"
我們傳參 flink run ... 無論怎么傳，只不過傳遞給了CliFrontend后面的 "$@" 而已。
因此 bin/flink 中 config.sh 讀取了 flink-conf.yaml最終存儲到JVM_ARGS 和 FLINK_ENV_JAVA_OPTS 才是真正的JVM參數。