You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ga...@apache.org on 2019/03/06 09:20:15 UTC
[flink] branch master updated (e2579e3 -> 6f84008)
This is an automated email from the ASF dual-hosted git repository.
gary pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from e2579e3 [FLINK-11827][table-runtime-blink] Introduce DataFormatConverters to convert between internal data format and java format
new 9ac14a6 [hotfix][runtime] Remove unused local variable in ExecutionGraph
new 6f84008 [FLINK-11781][yarn] Remove "DISABLED" as possible value for yarn.per-job-cluster.include-user-jar
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../generated/yarn_config_configuration.html | 2 +-
.../apache/flink/configuration/Configuration.java | 30 ++++++++++++
.../configuration/DelegatingConfiguration.java | 5 ++
.../flink/configuration/ConfigurationTest.java | 54 ++++++++++++++++++++++
.../runtime/executiongraph/ExecutionGraph.java | 1 -
.../flink/yarn/AbstractYarnClusterDescriptor.java | 40 ++++++++--------
.../yarn/configuration/YarnConfigOptions.java | 4 +-
.../flink/yarn/YarnClusterDescriptorTest.java | 23 +++++++++
8 files changed, 132 insertions(+), 27 deletions(-)
[flink] 02/02: [FLINK-11781][yarn] Remove "DISABLED" as possible
value for yarn.per-job-cluster.include-user-jar
Posted by ga...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gary pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 6f840082144c8a340a1ce0a820d268510e2853a1
Author: Gary Yao <ga...@apache.org>
AuthorDate: Sun Mar 3 19:12:54 2019 +0100
[FLINK-11781][yarn] Remove "DISABLED" as possible value for yarn.per-job-cluster.include-user-jar
Remove this feature because it is broken since Flink 1.5
This closes #7883.
---
.../generated/yarn_config_configuration.html | 2 +-
.../apache/flink/configuration/Configuration.java | 30 ++++++++++++
.../configuration/DelegatingConfiguration.java | 5 ++
.../flink/configuration/ConfigurationTest.java | 54 ++++++++++++++++++++++
.../flink/yarn/AbstractYarnClusterDescriptor.java | 40 ++++++++--------
.../yarn/configuration/YarnConfigOptions.java | 4 +-
.../flink/yarn/YarnClusterDescriptorTest.java | 23 +++++++++
7 files changed, 132 insertions(+), 26 deletions(-)
diff --git a/docs/_includes/generated/yarn_config_configuration.html b/docs/_includes/generated/yarn_config_configuration.html
index bbe2549..ab7e224 100644
--- a/docs/_includes/generated/yarn_config_configuration.html
+++ b/docs/_includes/generated/yarn_config_configuration.html
@@ -45,7 +45,7 @@
<tr>
<td><h5>yarn.per-job-cluster.include-user-jar</h5></td>
<td style="word-wrap: break-word;">"ORDER"</td>
- <td>Defines whether user-jars are included in the system class path for per-job-clusters as well as their positioning in the path. They can be positioned at the beginning ("FIRST"), at the end ("LAST"), or be positioned based on their name ("ORDER"). Setting this parameter to "DISABLED" causes the jar to be included in the user class path instead.</td>
+ <td>Defines whether user-jars are included in the system class path for per-job-clusters as well as their positioning in the path. They can be positioned at the beginning ("FIRST"), at the end ("LAST"), or be positioned based on their name ("ORDER").</td>
</tr>
<tr>
<td><h5>yarn.properties-file.location</h5></td>
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/Configuration.java b/flink-core/src/main/java/org/apache/flink/configuration/Configuration.java
index 07c0290..186cf10 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/Configuration.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/Configuration.java
@@ -33,10 +33,13 @@ import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.Locale;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
/**
* Lightweight configuration object which stores key/value pairs.
*/
@@ -608,6 +611,33 @@ public class Configuration extends ExecutionConfig.GlobalJobParameters
return o == null ? null : o.toString();
}
+ /**
+ * Returns the value associated with the given config option as an enum.
+ *
+ * @param enumClass The return enum class
+ * @param configOption The configuration option
+ * @throws IllegalArgumentException If the string associated with the given config option cannot
+ * be parsed as a value of the provided enum class.
+ */
+ @PublicEvolving
+ public <T extends Enum<T>> T getEnum(
+ final Class<T> enumClass,
+ final ConfigOption<String> configOption) {
+ checkNotNull(enumClass, "enumClass must not be null");
+ checkNotNull(configOption, "configOption must not be null");
+
+ final String configValue = getString(configOption);
+ try {
+ return Enum.valueOf(enumClass, configValue.toUpperCase(Locale.ROOT));
+ } catch (final IllegalArgumentException | NullPointerException e) {
+ final String errorMessage = String.format("Value for config option %s must be one of %s (was %s)",
+ configOption.key(),
+ Arrays.toString(enumClass.getEnumConstants()),
+ configValue);
+ throw new IllegalArgumentException(errorMessage, e);
+ }
+ }
+
// --------------------------------------------------------------------------------------------
/**
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/DelegatingConfiguration.java b/flink-core/src/main/java/org/apache/flink/configuration/DelegatingConfiguration.java
index 0a0a777..b0249a0 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/DelegatingConfiguration.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/DelegatingConfiguration.java
@@ -245,6 +245,11 @@ public final class DelegatingConfiguration extends Configuration {
}
@Override
+ public <T extends Enum<T>> T getEnum(final Class<T> enumClass, final ConfigOption<String> configOption) {
+ return this.backingConfig.getEnum(enumClass, prefixOption(configOption, prefix));
+ }
+
+ @Override
public void addAllToProperties(Properties props) {
// only add keys with our prefix
synchronized (backingConfig.confData) {
diff --git a/flink-core/src/test/java/org/apache/flink/configuration/ConfigurationTest.java b/flink-core/src/test/java/org/apache/flink/configuration/ConfigurationTest.java
index f02d59e..9727b44 100644
--- a/flink-core/src/test/java/org/apache/flink/configuration/ConfigurationTest.java
+++ b/flink-core/src/test/java/org/apache/flink/configuration/ConfigurationTest.java
@@ -23,10 +23,12 @@ import org.apache.flink.util.TestLogger;
import org.junit.Test;
+import static org.hamcrest.Matchers.containsString;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -403,4 +405,56 @@ public class ConfigurationTest extends TestLogger {
assertEquals("Wrong expectation about size", cfg.keySet().size(), 0);
assertFalse("Expected 'unexistedOption' is not removed", cfg.removeConfig(unexistedOption));
}
+
+ @Test
+ public void testShouldParseValidStringToEnum() {
+ final ConfigOption<String> configOption = createStringConfigOption();
+
+ final Configuration configuration = new Configuration();
+ configuration.setString(configOption.key(), TestEnum.VALUE1.toString());
+
+ final TestEnum parsedEnumValue = configuration.getEnum(TestEnum.class, configOption);
+ assertEquals(TestEnum.VALUE1, parsedEnumValue);
+ }
+
+ @Test
+ public void testShouldParseValidStringToEnumIgnoringCase() {
+ final ConfigOption<String> configOption = createStringConfigOption();
+
+ final Configuration configuration = new Configuration();
+ configuration.setString(configOption.key(), TestEnum.VALUE1.toString().toLowerCase());
+
+ final TestEnum parsedEnumValue = configuration.getEnum(TestEnum.class, configOption);
+ assertEquals(TestEnum.VALUE1, parsedEnumValue);
+ }
+
+ @Test
+ public void testThrowsExceptionIfTryingToParseInvalidStringForEnum() {
+ final ConfigOption<String> configOption = createStringConfigOption();
+
+ final Configuration configuration = new Configuration();
+ final String invalidValueForTestEnum = "InvalidValueForTestEnum";
+ configuration.setString(configOption.key(), invalidValueForTestEnum);
+
+ try {
+ configuration.getEnum(TestEnum.class, configOption);
+ fail("Expected exception not thrown");
+ } catch (IllegalArgumentException e) {
+ final String expectedMessage = "Value for config option " +
+ configOption.key() + " must be one of [VALUE1, VALUE2] (was " +
+ invalidValueForTestEnum + ")";
+ assertThat(e.getMessage(), containsString(expectedMessage));
+ }
+ }
+
+ enum TestEnum {
+ VALUE1,
+ VALUE2
+ }
+
+ private static ConfigOption<String> createStringConfigOption() {
+ return ConfigOptions
+ .key("test-string-key")
+ .noDefaultValue();
+ }
}
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java b/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
index a87772f..783a459 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
@@ -763,19 +763,14 @@ public abstract class AbstractYarnClusterDescriptor implements ClusterDescriptor
localResources,
envShipFileList);
- List<String> userClassPaths;
- if (userJarInclusion != YarnConfigOptions.UserJarInclusion.DISABLED) {
- userClassPaths = uploadAndRegisterFiles(
- userJarFiles,
- fs,
- homeDir,
- appId,
- paths,
- localResources,
- envShipFileList);
- } else {
- userClassPaths = Collections.emptyList();
- }
+ final List<String> userClassPaths = uploadAndRegisterFiles(
+ userJarFiles,
+ fs,
+ homeDir,
+ appId,
+ paths,
+ localResources,
+ envShipFileList);
if (userJarInclusion == YarnConfigOptions.UserJarInclusion.ORDER) {
systemClassPaths.addAll(userClassPaths);
@@ -1602,15 +1597,16 @@ public abstract class AbstractYarnClusterDescriptor implements ClusterDescriptor
}
private static YarnConfigOptions.UserJarInclusion getUserJarInclusionMode(org.apache.flink.configuration.Configuration config) {
- String configuredUserJarInclusion = config.getString(YarnConfigOptions.CLASSPATH_INCLUDE_USER_JAR);
- try {
- return YarnConfigOptions.UserJarInclusion.valueOf(configuredUserJarInclusion.toUpperCase());
- } catch (IllegalArgumentException e) {
- LOG.warn("Configuration parameter {} was configured with an invalid value {}. Falling back to default ({}).",
- YarnConfigOptions.CLASSPATH_INCLUDE_USER_JAR.key(),
- configuredUserJarInclusion,
- YarnConfigOptions.CLASSPATH_INCLUDE_USER_JAR.defaultValue());
- return YarnConfigOptions.UserJarInclusion.valueOf(YarnConfigOptions.CLASSPATH_INCLUDE_USER_JAR.defaultValue());
+ throwIfUserTriesToDisableUserJarInclusionInSystemClassPath(config);
+
+ return config.getEnum(YarnConfigOptions.UserJarInclusion.class, YarnConfigOptions.CLASSPATH_INCLUDE_USER_JAR);
+ }
+
+ private static void throwIfUserTriesToDisableUserJarInclusionInSystemClassPath(final Configuration config) {
+ final String userJarInclusion = config.getString(YarnConfigOptions.CLASSPATH_INCLUDE_USER_JAR);
+ if ("DISABLED".equalsIgnoreCase(userJarInclusion)) {
+ throw new IllegalArgumentException(String.format("Config option %s cannot be set to DISABLED anymore (see FLINK-11781)",
+ YarnConfigOptions.CLASSPATH_INCLUDE_USER_JAR.key()));
}
}
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/configuration/YarnConfigOptions.java b/flink-yarn/src/main/java/org/apache/flink/yarn/configuration/YarnConfigOptions.java
index b059475..6abeb0d 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/configuration/YarnConfigOptions.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/configuration/YarnConfigOptions.java
@@ -57,8 +57,7 @@ public class YarnConfigOptions {
.defaultValue("ORDER")
.withDescription("Defines whether user-jars are included in the system class path for per-job-clusters as" +
" well as their positioning in the path. They can be positioned at the beginning (\"FIRST\"), at the" +
- " end (\"LAST\"), or be positioned based on their name (\"ORDER\"). Setting this parameter to" +
- " \"DISABLED\" causes the jar to be included in the user class path instead.");
+ " end (\"LAST\"), or be positioned based on their name (\"ORDER\").");
/**
* The vcores exposed by YARN.
@@ -156,7 +155,6 @@ public class YarnConfigOptions {
/** @see YarnConfigOptions#CLASSPATH_INCLUDE_USER_JAR */
public enum UserJarInclusion {
- DISABLED,
FIRST,
LAST,
ORDER
diff --git a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java
index 16b3586..6c2e42e 100644
--- a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java
+++ b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java
@@ -53,7 +53,9 @@ import java.util.Map;
import java.util.Set;
import static junit.framework.TestCase.assertTrue;
+import static org.hamcrest.Matchers.containsString;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;
/**
@@ -89,6 +91,27 @@ public class YarnClusterDescriptorTest extends TestLogger {
yarnClient.stop();
}
+ /**
+ * @see <a href="https://issues.apache.org/jira/browse/FLINK-11781">FLINK-11781</a>
+ */
+ @Test
+ public void testThrowsExceptionIfUserTriesToDisableUserJarInclusionInSystemClassPath() {
+ final Configuration configuration = new Configuration();
+ configuration.setString(YarnConfigOptions.CLASSPATH_INCLUDE_USER_JAR, "DISABLED");
+
+ try {
+ new YarnClusterDescriptor(
+ configuration,
+ yarnConfiguration,
+ temporaryFolder.getRoot().getAbsolutePath(),
+ yarnClient,
+ true);
+ fail("Expected exception not thrown");
+ } catch (final IllegalArgumentException e) {
+ assertThat(e.getMessage(), containsString("cannot be set to DISABLED anymore"));
+ }
+ }
+
@Test
public void testFailIfTaskSlotsHigherThanMaxVcores() throws ClusterDeploymentException {
final Configuration flinkConfiguration = new Configuration();
[flink] 01/02: [hotfix][runtime] Remove unused local variable in
ExecutionGraph
Posted by ga...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gary pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 9ac14a6048982d5bdd19379114ef109483c2abce
Author: Gary Yao <ga...@apache.org>
AuthorDate: Fri Mar 1 18:06:38 2019 +0100
[hotfix][runtime] Remove unused local variable in ExecutionGraph
---
.../java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java | 1 -
1 file changed, 1 deletion(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
index e64b2b9..ff7e123 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
@@ -1825,7 +1825,6 @@ public class ExecutionGraph implements AccessExecutionGraph {
// see what this means for us. currently, the first FAILED state means -> FAILED
if (newExecutionState == ExecutionState.FAILED) {
final Throwable ex = error != null ? error : new FlinkException("Unknown Error (missing cause)");
- long timestamp = execution.getStateTimestamp(ExecutionState.FAILED);
// by filtering out late failure calls, we can save some work in
// avoiding redundant local failover