zjffdu commented on a change in pull request #4116:
URL: https://github.com/apache/zeppelin/pull/4116#discussion_r633565472
##########
File path: flink/interpreter/pom.xml
##########
@@ -95,6 +96,13 @@
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>io.fabric8</groupId>
+ <artifactId>kubernetes-client</artifactId>
+ <version>${kubernetes.client.version}</version>
+ <scope>compile</scope>
Review comment:
Right，I didn't use it yet, Just plan to add ingress support in next PR.
Let me remove it in this PR.
##########
File path:
zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServer.java
##########
@@ -142,6 +141,7 @@
private ScheduledExecutorService resultCleanService =
Executors.newSingleThreadScheduledExecutor();
private boolean isTest;
+ private boolean isFlinkK8sApplicationMode = false;
Review comment:
That's right, but I think this is the only way for flink k8s mode. For
flink k8s mode, the `RemoteInterpreterServer#main` is the not the actual main
entry point, `RemoteInterpreterServer#main` will be called by flink framework
via reflection. So if we call `System.exit()` method in flink's k8s mode, it
would make flink framework think that the flink job is exited abnormally, and
will restart it again which is not what we expect. Let me add some comment to
explain it in this class
##########
File path:
flink/interpreter/src/main/scala/org/apache/zeppelin/flink/FlinkScalaInterpreter.scala
##########
@@ -312,9 +323,12 @@ class FlinkScalaInterpreter(val properties: Properties) {
// remote mode
if (mode == ExecutionMode.YARN_APPLICATION) {
val yarnAppId = System.getenv("_APP_ID");
- LOGGER.info("Use FlinkCluster in yarn application mode, appId:
{}", yarnAppId)
+ LOGGER.info("Use FlinkCluster in yarn-application mode, appId:
{}", yarnAppId)
this.jmWebUrl = "http://localhost:"; +
HadoopUtils.getFlinkRestPort(yarnAppId)
this.displayedJMWebUrl =
HadoopUtils.getYarnAppTrackingUrl(yarnAppId)
+ } else if (mode == ExecutionMode.KUBERNETES_APPLICATION) {
+ LOGGER.info("Use FlinkCluster in kubernetes-application mode")
+ this.jmWebUrl = "http://localhost:8083";
} else {
Review comment:
I split it into 2 methods: createFlinkILoop & initFlinkILoop
##########
File path:
zeppelin-plugins/launcher/flink/src/main/java/org/apache/zeppelin/interpreter/launcher/FlinkInterpreterLauncher.java
##########
@@ -45,25 +48,44 @@ public FlinkInterpreterLauncher(ZeppelinConfiguration
zConf, RecoveryStorage rec
throws IOException {
Map<String, String> envs = super.buildEnvFromProperties(context);
- String flinkHome = updateEnvsForFlinkHome(envs, context);
-
+ String flinkHome = getFlinkHome(context);
if (!envs.containsKey("FLINK_CONF_DIR")) {
envs.put("FLINK_CONF_DIR", flinkHome + "/conf");
}
envs.put("FLINK_LIB_DIR", flinkHome + "/lib");
envs.put("FLINK_PLUGINS_DIR", flinkHome + "/plugins");
- // yarn application mode specific logic
- if ("yarn-application".equalsIgnoreCase(
- context.getProperties().getProperty("flink.execution.mode"))) {
- updateEnvsForYarnApplicationMode(envs, context);
+ String mode = context.getProperties().getProperty("flink.execution.mode");
+ String a = FLINK_EXECUTION_MODES.stream().collect(Collectors.joining(",
"));
Review comment:
Good catch, let me remove it
##########
File path:
zeppelin-plugins/launcher/flink/src/main/java/org/apache/zeppelin/interpreter/launcher/FlinkInterpreterLauncher.java
##########
@@ -45,25 +48,44 @@ public FlinkInterpreterLauncher(ZeppelinConfiguration
zConf, RecoveryStorage rec
throws IOException {
Map<String, String> envs = super.buildEnvFromProperties(context);
- String flinkHome = updateEnvsForFlinkHome(envs, context);
-
+ String flinkHome = getFlinkHome(context);
if (!envs.containsKey("FLINK_CONF_DIR")) {
envs.put("FLINK_CONF_DIR", flinkHome + "/conf");
}
envs.put("FLINK_LIB_DIR", flinkHome + "/lib");
envs.put("FLINK_PLUGINS_DIR", flinkHome + "/plugins");
- // yarn application mode specific logic
- if ("yarn-application".equalsIgnoreCase(
- context.getProperties().getProperty("flink.execution.mode"))) {
- updateEnvsForYarnApplicationMode(envs, context);
+ String mode = context.getProperties().getProperty("flink.execution.mode");
+ String a = FLINK_EXECUTION_MODES.stream().collect(Collectors.joining(",
"));
+ if (!FLINK_EXECUTION_MODES.contains(mode)) {
+ throw new IOException("Not valid flink.execution.mode: " +
+ mode + ", valid modes ares: " +
+ FLINK_EXECUTION_MODES.stream().collect(Collectors.joining(",
")));
+ }
+ if (isApplicationMode(mode)) {
+ updateEnvsForApplicationMode(mode, envs, context);
}
return envs;
}
- private String updateEnvsForFlinkHome(Map<String, String> envs,
- InterpreterLaunchContext context) throws
IOException {
+ private void verifyExecutionMode(String mode) {
+
+ }
Review comment:
removed
##########
File path:
zeppelin-plugins/launcher/flink/src/main/java/org/apache/zeppelin/interpreter/launcher/FlinkInterpreterLauncher.java
##########
@@ -45,25 +48,44 @@ public FlinkInterpreterLauncher(ZeppelinConfiguration
zConf, RecoveryStorage rec
throws IOException {
Map<String, String> envs = super.buildEnvFromProperties(context);
- String flinkHome = updateEnvsForFlinkHome(envs, context);
-
+ String flinkHome = getFlinkHome(context);
if (!envs.containsKey("FLINK_CONF_DIR")) {
envs.put("FLINK_CONF_DIR", flinkHome + "/conf");
}
envs.put("FLINK_LIB_DIR", flinkHome + "/lib");
envs.put("FLINK_PLUGINS_DIR", flinkHome + "/plugins");
- // yarn application mode specific logic
- if ("yarn-application".equalsIgnoreCase(
- context.getProperties().getProperty("flink.execution.mode"))) {
- updateEnvsForYarnApplicationMode(envs, context);
+ String mode = context.getProperties().getProperty("flink.execution.mode");
+ String a = FLINK_EXECUTION_MODES.stream().collect(Collectors.joining(",
"));
+ if (!FLINK_EXECUTION_MODES.contains(mode)) {
+ throw new IOException("Not valid flink.execution.mode: " +
+ mode + ", valid modes ares: " +
+ FLINK_EXECUTION_MODES.stream().collect(Collectors.joining(",
")));
+ }
+ if (isApplicationMode(mode)) {
+ updateEnvsForApplicationMode(mode, envs, context);
}
return envs;
}
- private String updateEnvsForFlinkHome(Map<String, String> envs,
- InterpreterLaunchContext context) throws
IOException {
+ private void verifyExecutionMode(String mode) {
+
+ }
+
+ private boolean isApplicationMode(String mode) {
+ return "yarn-application".equals(mode) ||
"kubernetes-application".equals(mode);
Review comment:
Fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
