[FLINK-15179] Assume Deployment.target=kubernetes-session when in Kub…

…ernetesSessionCli
mbalassi · Dec 14, 2019 · 2ec51f6 · 2ec51f6
1 parent bc8964c
commit 2ec51f6
Show file tree

Hide file tree

Showing 4 changed files with 15 additions and 14 deletions.
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/ExecutorCLI.java b/flink-clients/src/main/java/org/apache/flink/client/cli/ExecutorCLI.java
@@ -32,7 +32,6 @@
 import java.util.Properties;
 
 import static org.apache.flink.util.Preconditions.checkNotNull;
-import static org.apache.flink.util.Preconditions.checkState;
 
 /**
  * A generic implementation of the {@link CustomCommandLine} that only expects
@@ -92,8 +91,6 @@ public void addGeneralOptions(Options baseOptions) {
 
  @Override
  public Configuration applyCommandLineOptionsToConfiguration(final CommandLine commandLine) {
- checkState(isActive(commandLine));
-
  final Configuration effectiveConfiguration = new Configuration(baseConfiguration);
 
  final String executorName = commandLine.getOptionValue(executorOption.getOpt());

diff --git a/flink-clients/src/test/java/org/apache/flink/client/cli/ExecutorCLITest.java b/flink-clients/src/test/java/org/apache/flink/client/cli/ExecutorCLITest.java
@@ -51,16 +51,6 @@ public void initOptions() {
  cliUnderTest.addGeneralOptions(testOptions);
  }
 
- @Test(expected = IllegalStateException.class)
- public void testExceptionWithUnspecifiedExecutor() throws CliArgsException {
- final ExecutorCLI cliUnderTest = new ExecutorCLI(new Configuration());
-
- final String[] args = {"-Dtest=tsetD-"};
- final CommandLine commandLine = CliFrontendParser.parse(testOptions, args, true);
-
- cliUnderTest.applyCommandLineOptionsToConfiguration(commandLine);
- }
-
  @Test
  public void testExecutorInBaseConfigIsPickedUp() throws CliArgsException {
  final String expectedExecutorName = "test-executor";

diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/cli/KubernetesSessionCli.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/cli/KubernetesSessionCli.java
@@ -31,6 +31,7 @@
 import org.apache.flink.configuration.DeploymentOptions;
 import org.apache.flink.configuration.GlobalConfiguration;
 import org.apache.flink.configuration.UnmodifiableConfiguration;
+import org.apache.flink.kubernetes.executors.KubernetesSessionClusterExecutor;
 import org.apache.flink.kubernetes.kubeclient.FlinkKubeClient;
 import org.apache.flink.kubernetes.kubeclient.KubeClientFactory;
 import org.apache.flink.runtime.security.SecurityUtils;
@@ -77,7 +78,9 @@ public KubernetesSessionCli(Configuration configuration, ClusterClientServiceLoa
 
  public Configuration getEffectiveConfiguration(String[] args) throws CliArgsException {
  final CommandLine commandLine = cli.parseCommandLineOptions(args, true);
- return cli.applyCommandLineOptionsToConfiguration(commandLine);
+ final Configuration effectiveConfiguration = cli.applyCommandLineOptionsToConfiguration(commandLine);
+ effectiveConfiguration.set(DeploymentOptions.TARGET, KubernetesSessionClusterExecutor.NAME);
+ return effectiveConfiguration;
  }
 
  private int run(String[] args) throws FlinkException, CliArgsException {

diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesSessionCliTest.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesSessionCliTest.java
@@ -18,6 +18,7 @@
 
 package org.apache.flink.kubernetes;
 
+import org.apache.flink.client.cli.CliArgsException;
 import org.apache.flink.client.deployment.ClusterClientFactory;
 import org.apache.flink.client.deployment.ClusterClientServiceLoader;
 import org.apache.flink.client.deployment.ClusterSpecification;
@@ -45,6 +46,16 @@
  */
 public class KubernetesSessionCliTest {
 
+ @Test
+ public void testKubernetesSessionCliSetsDeploymentTargetCorrectly() throws CliArgsException {
+ final KubernetesSessionCli cli = new KubernetesSessionCli(new Configuration());
+
+ final String[] args = {};
+ final Configuration configuration = cli.getEffectiveConfiguration(args);
+
+ assertEquals(KubernetesSessionClusterExecutor.NAME, configuration.get(DeploymentOptions.TARGET));
+ }
+
  @Test
  public void testDynamicProperties() throws Exception {