[FLINK-31321][Deployment/YARN] Yarn-session mode, securityConfigurati…

…on supports dynamic configuration This closes apache#22119
PatrickRen · Mar 17, 2023 · 6f9bca9 · 6f9bca9
1 parent 81f77a9
commit 6f9bca9
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 3 deletions.
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/DynamicPropertiesUtil.java b/flink-clients/src/main/java/org/apache/flink/client/cli/DynamicPropertiesUtil.java
@@ -29,7 +29,7 @@
  * Helper class for supporting dynamic property commandline options in {@link CustomCommandLine
  * CustomCommandLines}.
  */
-class DynamicPropertiesUtil {
+public class DynamicPropertiesUtil {
 
     /**
      * Dynamic properties allow the user to specify additional configuration values with -D, such as
@@ -49,7 +49,7 @@ class DynamicPropertiesUtil {
      * Parses dynamic properties from the given {@link CommandLine} and sets them on the {@link
      * Configuration}.
      */
-    static void encodeDynamicProperties(
+    public static void encodeDynamicProperties(
             final CommandLine commandLine, final Configuration effectiveConfiguration) {
 
         final Properties properties = commandLine.getOptionProperties(DYNAMIC_PROPERTIES.getOpt());

diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java b/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
@@ -20,6 +20,8 @@
 
 import org.apache.flink.client.cli.CliArgsException;
 import org.apache.flink.client.cli.CliFrontend;
+import org.apache.flink.client.cli.CliFrontendParser;
+import org.apache.flink.client.cli.DynamicPropertiesUtil;
 import org.apache.flink.client.deployment.ClusterClientFactory;
 import org.apache.flink.client.deployment.ClusterClientServiceLoader;
 import org.apache.flink.client.deployment.ClusterSpecification;
@@ -853,7 +855,12 @@ public static void main(final String[] args) {
                             "",
                             ""); // no prefix for the YARN session
 
-            SecurityUtils.install(new SecurityConfiguration(flinkConfiguration));
+            final CommandLine commandLine = CliFrontendParser.parse(cli.allOptions, args, true);
+
+            final Configuration securityFlinkConfiguration = flinkConfiguration.clone();
+            DynamicPropertiesUtil.encodeDynamicProperties(commandLine, securityFlinkConfiguration);
+
+            SecurityUtils.install(new SecurityConfiguration(securityFlinkConfiguration));
 
             retCode = SecurityUtils.getInstalledContext().runSecured(() -> cli.run(args));
         } catch (CliArgsException e) {