Skip to content

Commit

Permalink
[FLINK-11534] [container] Allow ExecutionMode configuration
Browse files Browse the repository at this point in the history
  • Loading branch information
uce committed Apr 24, 2019
1 parent c5d2a57 commit 882efcb
Show file tree
Hide file tree
Showing 2 changed files with 83 additions and 2 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

package org.apache.flink.container.entrypoint;

import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.JobID;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.entrypoint.ClusterEntrypoint;
Expand Down Expand Up @@ -92,8 +93,7 @@ public static void main(String[] args) {
}

Configuration configuration = loadConfiguration(clusterConfiguration);

configuration.setString(ClusterEntrypoint.EXECUTION_MODE, ExecutionMode.DETACHED.toString());
setDefaultExecutionModeIfNotConfigured(configuration);

StandaloneJobClusterEntryPoint entrypoint = new StandaloneJobClusterEntryPoint(
configuration,
Expand All @@ -104,4 +104,16 @@ public static void main(String[] args) {

ClusterEntrypoint.runClusterEntrypoint(entrypoint);
}

@VisibleForTesting
static void setDefaultExecutionModeIfNotConfigured(Configuration configuration) {
if (isNoExecutionModeConfigured(configuration)) {
// In contrast to other places, the default for standalone job clusters is ExecutionMode.DETACHED
configuration.setString(ClusterEntrypoint.EXECUTION_MODE, ExecutionMode.DETACHED.toString());
}
}

private static boolean isNoExecutionModeConfigured(Configuration configuration) {
return configuration.getString(ClusterEntrypoint.EXECUTION_MODE, null) == null;
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,69 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package org.apache.flink.container.entrypoint;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.entrypoint.ClusterEntrypoint;
import org.apache.flink.runtime.entrypoint.ClusterEntrypoint.ExecutionMode;
import org.apache.flink.util.TestLogger;

import org.junit.Test;

import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.MatcherAssert.assertThat;

/**
* Tests for the {@link StandaloneJobClusterEntryPoint}.
*/
public class StandaloneJobClusterEntryPointTest extends TestLogger {

/**
* Tests that the default {@link ExecutionMode} is {@link ExecutionMode#DETACHED}.
*/
@Test
public void testDefaultExecutionModeIsDetached() {
Configuration configuration = new Configuration();

StandaloneJobClusterEntryPoint.setDefaultExecutionModeIfNotConfigured(configuration);

assertThat(getExecutionMode(configuration), equalTo(ExecutionMode.DETACHED));
}

/**
* Tests that {@link ExecutionMode} is not overwritten if provided.
*/
@Test
public void testDontOverwriteExecutionMode() {
Configuration configuration = new Configuration();
setExecutionMode(configuration, ExecutionMode.NORMAL);

StandaloneJobClusterEntryPoint.setDefaultExecutionModeIfNotConfigured(configuration);

// Don't overwrite provided configuration
assertThat(getExecutionMode(configuration), equalTo(ExecutionMode.NORMAL));
}

private static void setExecutionMode(Configuration configuration, ExecutionMode executionMode) {
configuration.setString(ClusterEntrypoint.EXECUTION_MODE, executionMode.toString());
}

private static ExecutionMode getExecutionMode(Configuration configuration) {
return ExecutionMode.valueOf(configuration.getString(ClusterEntrypoint.EXECUTION_MODE));
}
}

0 comments on commit 882efcb

Please sign in to comment.