[FLINK-37638][clients] Avoid duplicate call on getEffectiveConfiguration

Author: beliefer

Diff for: flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java

@@ -187,32 +187,28 @@ protected void run(String[] args) throws Exception {
         final CustomCommandLine activeCommandLine =
                 validateAndGetActiveCommandLine(checkNotNull(commandLine));
 
-        if (isDeploymentTargetApplication(activeCommandLine, commandLine)) {
+        final Configuration effectiveConfiguration =
+                getEffectiveConfiguration(activeCommandLine, commandLine);
+
+        if (isDeploymentTargetApplication(effectiveConfiguration)) {
             final ApplicationDeployer deployer =
                     new ApplicationClusterDeployer(clusterClientServiceLoader);
 
             final ProgramOptions programOptions;
-            final Configuration effectiveConfiguration;
 
             // No need to set a jarFile path for PyFlink job.
             if (ProgramOptionsUtils.isPythonEntryPoint(commandLine)) {
                 programOptions = ProgramOptionsUtils.createPythonProgramOptions(commandLine);
-                effectiveConfiguration =
-                        getEffectiveConfiguration(
-                                activeCommandLine,
-                                commandLine,
-                                programOptions,
-                                Collections.emptyList());
+                updateEffectiveConfiguration(
+                        effectiveConfiguration, programOptions, Collections.emptyList());
             } else {
                 programOptions = new ProgramOptions(commandLine);
                 programOptions.validate();
                 final URI uri = PackagedProgramUtils.resolveURI(programOptions.getJarFilePath());
-                effectiveConfiguration =
-                        getEffectiveConfiguration(
-                                activeCommandLine,
-                                commandLine,
-                                programOptions,
-                                Collections.singletonList(uri.toString()));
+                updateEffectiveConfiguration(
+                        effectiveConfiguration,
+                        programOptions,
+                        Collections.singletonList(uri.toString()));
             }
 
             final ApplicationConfiguration applicationConfiguration =
@@ -225,9 +221,7 @@ protected void run(String[] args) throws Exception {
 
             final List<URL> jobJars = getJobJarAndDependencies(programOptions);
 
-            final Configuration effectiveConfiguration =
-                    getEffectiveConfiguration(
-                            activeCommandLine, commandLine, programOptions, jobJars);
+            updateEffectiveConfiguration(effectiveConfiguration, programOptions, jobJars);
 
             LOG.debug("Effective executor configuration: {}", effectiveConfiguration);
 
@@ -238,12 +232,7 @@ protected void run(String[] args) throws Exception {
         }
     }
 
-    protected boolean isDeploymentTargetApplication(
-            final CustomCommandLine activeCustomCommandLine, final CommandLine commandLine)
-            throws FlinkException {
-        final Configuration effectiveConfiguration =
-                getEffectiveConfiguration(activeCustomCommandLine, commandLine);
-
+    protected boolean isDeploymentTargetApplication(final Configuration effectiveConfiguration) {
         final String executionTarget =
                 effectiveConfiguration
                         .getOptional(DeploymentOptions.TARGET)
@@ -300,15 +289,10 @@ private Configuration getEffectiveConfiguration(
         return effectiveConfiguration;
     }
 
-    private <T> Configuration getEffectiveConfiguration(
-            final CustomCommandLine activeCustomCommandLine,
-            final CommandLine commandLine,
+    private <T> void updateEffectiveConfiguration(
+            final Configuration effectiveConfiguration,
             final ProgramOptions programOptions,
-            final List<T> jobJars)
-            throws FlinkException {
-
-        final Configuration effectiveConfiguration =
-                getEffectiveConfiguration(activeCustomCommandLine, commandLine);
+            final List<T> jobJars) {
 
         final ExecutionConfigAccessor executionParameters =
                 ExecutionConfigAccessor.fromProgramOptions(programOptions, jobJars);
@@ -318,7 +302,6 @@ private <T> Configuration getEffectiveConfiguration(
         LOG.debug(
                 "Effective configuration after Flink conf, custom commandline, and program options: {}",
                 effectiveConfiguration);
-        return effectiveConfiguration;
     }
 
     /**
@@ -354,11 +337,12 @@ protected void info(String[] args) throws Exception {
                     validateAndGetActiveCommandLine(checkNotNull(commandLine));
 
             final Configuration effectiveConfiguration =
-                    getEffectiveConfiguration(
-                            activeCommandLine,
-                            commandLine,
-                            programOptions,
-                            getJobJarAndDependencies(programOptions));
+                    getEffectiveConfiguration(activeCommandLine, commandLine);
+
+            updateEffectiveConfiguration(
+                    effectiveConfiguration,
+                    programOptions,
+                    getJobJarAndDependencies(programOptions));
 
             program = buildProgram(programOptions, effectiveConfiguration);
 

Diff for: flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendRunTest.java

@@ -28,7 +28,6 @@
 import org.apache.flink.configuration.CoreOptions;
 import org.apache.flink.core.execution.RecoveryClaimMode;
 import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
-import org.apache.flink.util.FlinkException;
 
 import org.apache.commons.cli.CommandLine;
 import org.junit.jupiter.api.AfterAll;
@@ -376,10 +375,8 @@ protected void run(String[] args) throws Exception {
         }
 
         @Override
-        protected boolean isDeploymentTargetApplication(
-                CustomCommandLine activeCustomCommandLine, CommandLine commandLine)
-                throws FlinkException {
-            application = super.isDeploymentTargetApplication(activeCustomCommandLine, commandLine);
+        protected boolean isDeploymentTargetApplication(Configuration effectiveConfiguration) {
+            application = super.isDeploymentTargetApplication(effectiveConfiguration);
             return application;
         }