You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by kk...@apache.org on 2020/04/30 20:10:55 UTC
[flink] 11/13: [minor] Move getUsrLibDir() to YarnEntrypointUtils
This is an automated email from the ASF dual-hosted git repository.
kkloudas pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 9e8ad72d672a87faf267d2da26ca7e70f3e8027d
Author: Kostas Kloudas <kk...@gmail.com>
AuthorDate: Tue Apr 28 23:53:54 2020 +0200
[minor] Move getUsrLibDir() to YarnEntrypointUtils
---
.../YarnApplicationClusterEntryPoint.java | 22 +-----------------
.../flink/yarn/entrypoint/YarnEntrypointUtils.java | 20 ++++++++++++++++
.../yarn/entrypoint/YarnJobClusterEntrypoint.java | 27 +++-------------------
3 files changed, 24 insertions(+), 45 deletions(-)
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnApplicationClusterEntryPoint.java b/flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnApplicationClusterEntryPoint.java
index 8dec082..dd1ce3b 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnApplicationClusterEntryPoint.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnApplicationClusterEntryPoint.java
@@ -35,20 +35,14 @@ import org.apache.flink.runtime.util.JvmShutdownSafeguard;
import org.apache.flink.runtime.util.SignalHandler;
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.Preconditions;
-import org.apache.flink.yarn.configuration.YarnConfigOptions;
import org.apache.hadoop.yarn.api.ApplicationConstants;
import javax.annotation.Nullable;
-import java.io.File;
import java.io.IOException;
import java.net.URL;
import java.util.Map;
-import java.util.Optional;
-
-import static org.apache.flink.runtime.util.ClusterEntrypointUtils.tryFindUserLibDirectory;
-import static org.apache.flink.util.Preconditions.checkState;
/**
* An {@link ApplicationClusterEntryPoint} for Yarn.
@@ -122,21 +116,7 @@ public final class YarnApplicationClusterEntryPoint extends ApplicationClusterEn
ClassPathPackagedProgramRetriever
.newBuilder(programArguments)
.setJobClassName(jobClassName);
- getUsrLibDir(configuration).ifPresent(retrieverBuilder::setUserLibDirectory);
+ YarnEntrypointUtils.getUsrLibDir(configuration).ifPresent(retrieverBuilder::setUserLibDirectory);
return retrieverBuilder.build();
}
-
- private static Optional<File> getUsrLibDir(final Configuration configuration) {
- final YarnConfigOptions.UserJarInclusion userJarInclusion = configuration
- .getEnum(YarnConfigOptions.UserJarInclusion.class, YarnConfigOptions.CLASSPATH_INCLUDE_USER_JAR);
- final Optional<File> userLibDir = tryFindUserLibDirectory();
-
- checkState(
- userJarInclusion != YarnConfigOptions.UserJarInclusion.DISABLED || userLibDir.isPresent(),
- "The %s is set to %s. But the usrlib directory does not exist.",
- YarnConfigOptions.CLASSPATH_INCLUDE_USER_JAR.key(),
- YarnConfigOptions.UserJarInclusion.DISABLED);
-
- return userJarInclusion == YarnConfigOptions.UserJarInclusion.DISABLED ? userLibDir : Optional.empty();
- }
}
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnEntrypointUtils.java b/flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnEntrypointUtils.java
index 3da17bb..c0887e1 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnEntrypointUtils.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnEntrypointUtils.java
@@ -31,13 +31,19 @@ import org.apache.flink.runtime.clusterframework.BootstrapTools;
import org.apache.flink.util.Preconditions;
import org.apache.flink.yarn.Utils;
import org.apache.flink.yarn.YarnConfigKeys;
+import org.apache.flink.yarn.configuration.YarnConfigOptions;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.yarn.api.ApplicationConstants;
import org.slf4j.Logger;
+import java.io.File;
import java.io.IOException;
import java.util.Map;
+import java.util.Optional;
+
+import static org.apache.flink.runtime.util.ClusterEntrypointUtils.tryFindUserLibDirectory;
+import static org.apache.flink.util.Preconditions.checkState;
/**
* This class contains utility methods for the {@link YarnSessionClusterEntrypoint} and
@@ -112,4 +118,18 @@ public class YarnEntrypointUtils {
log.info("YARN daemon is running as: {} Yarn client user obtainer: {}",
currentUser.getShortUserName(), yarnClientUsername);
}
+
+ public static Optional<File> getUsrLibDir(final Configuration configuration) {
+ final YarnConfigOptions.UserJarInclusion userJarInclusion = configuration
+ .getEnum(YarnConfigOptions.UserJarInclusion.class, YarnConfigOptions.CLASSPATH_INCLUDE_USER_JAR);
+ final Optional<File> userLibDir = tryFindUserLibDirectory();
+
+ checkState(
+ userJarInclusion != YarnConfigOptions.UserJarInclusion.DISABLED || userLibDir.isPresent(),
+ "The %s is set to %s. But the usrlib directory does not exist.",
+ YarnConfigOptions.CLASSPATH_INCLUDE_USER_JAR.key(),
+ YarnConfigOptions.UserJarInclusion.DISABLED);
+
+ return userJarInclusion == YarnConfigOptions.UserJarInclusion.DISABLED ? userLibDir : Optional.empty();
+ }
}
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnJobClusterEntrypoint.java b/flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnJobClusterEntrypoint.java
index d14ef05..4fdffc3 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnJobClusterEntrypoint.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnJobClusterEntrypoint.java
@@ -31,15 +31,8 @@ import org.apache.flink.yarn.configuration.YarnConfigOptions;
import org.apache.hadoop.yarn.api.ApplicationConstants;
-import javax.annotation.Nullable;
-
-import java.io.File;
import java.io.IOException;
import java.util.Map;
-import java.util.Optional;
-
-import static org.apache.flink.runtime.util.ClusterEntrypointUtils.tryFindUserLibDirectory;
-import static org.apache.flink.util.Preconditions.checkState;
/**
* Entry point for Yarn per-job clusters.
@@ -59,25 +52,11 @@ public class YarnJobClusterEntrypoint extends JobClusterEntrypoint {
protected DefaultDispatcherResourceManagerComponentFactory createDispatcherResourceManagerComponentFactory(Configuration configuration) throws IOException {
return DefaultDispatcherResourceManagerComponentFactory.createJobComponentFactory(
YarnResourceManagerFactory.getInstance(),
- FileJobGraphRetriever.createFrom(configuration, getUsrLibDir(configuration)));
+ FileJobGraphRetriever.createFrom(
+ configuration,
+ YarnEntrypointUtils.getUsrLibDir(configuration).orElse(null)));
}
- @Nullable
- private static File getUsrLibDir(final Configuration configuration) {
- final YarnConfigOptions.UserJarInclusion userJarInclusion = configuration
- .getEnum(YarnConfigOptions.UserJarInclusion.class, YarnConfigOptions.CLASSPATH_INCLUDE_USER_JAR);
- final Optional<File> userLibDir = tryFindUserLibDirectory();
-
- checkState(
- userJarInclusion != YarnConfigOptions.UserJarInclusion.DISABLED || userLibDir.isPresent(),
- "The %s is set to %s. But the usrlib directory does not exist.",
- YarnConfigOptions.CLASSPATH_INCLUDE_USER_JAR.key(),
- YarnConfigOptions.UserJarInclusion.DISABLED);
-
- return userJarInclusion == YarnConfigOptions.UserJarInclusion.DISABLED ? userLibDir.get() : null;
- }
-
-
// ------------------------------------------------------------------------
// The executable entry point for the Yarn Application Master Process
// for a single Flink job.