You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by mx...@apache.org on 2016/08/29 16:50:10 UTC
[1/3] flink git commit: [FLINK-4488][yarn] only automatically
shutdown clusters for detached jobs
Repository: flink
Updated Branches:
refs/heads/release-1.1 3cf494a3e -> f7ec1efb3
[FLINK-4488][yarn] only automatically shutdown clusters for detached jobs
- add check to yarn tests to verify cluster hasn't been shutdown prematurely
This closes #2419
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/28da9954
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/28da9954
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/28da9954
Branch: refs/heads/release-1.1
Commit: 28da995494ad21223f6911f27eb46187294f311a
Parents: 3cf494a
Author: Maximilian Michels <mx...@apache.org>
Authored: Wed Aug 17 15:41:12 2016 +0200
Committer: Maximilian Michels <mx...@apache.org>
Committed: Mon Aug 29 18:18:56 2016 +0200
----------------------------------------------------------------------
.../flink/yarn/FlinkYarnSessionCliTest.java | 2 +-
.../org/apache/flink/yarn/YarnTestBase.java | 92 ++++++++++++++++----
.../apache/flink/yarn/YarnClusterClient.java | 10 +--
3 files changed, 79 insertions(+), 25 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/28da9954/flink-yarn-tests/src/test/java/org/apache/flink/yarn/FlinkYarnSessionCliTest.java
----------------------------------------------------------------------
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/FlinkYarnSessionCliTest.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/FlinkYarnSessionCliTest.java
index 510a048..d03d9eb 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/FlinkYarnSessionCliTest.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/FlinkYarnSessionCliTest.java
@@ -150,7 +150,7 @@ public class FlinkYarnSessionCliTest {
FlinkYarnSessionCli yarnCLI = new TestCLI("y", "yarn");
AbstractYarnClusterDescriptor descriptor = yarnCLI.createDescriptor("", runOptions.getCommandLine());
- System.out.println(descriptor.getZookeeperNamespace());
+
Assert.assertEquals(zkNamespaceCliInput, descriptor.getZookeeperNamespace());
}
http://git-wip-us.apache.org/repos/asf/flink/blob/28da9954/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java
----------------------------------------------------------------------
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java
index 76b5d31..0243012 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java
@@ -18,9 +18,13 @@
package org.apache.flink.yarn;
+import akka.actor.Identify;
import org.apache.commons.io.FileUtils;
import org.apache.flink.client.CliFrontend;
+import org.apache.flink.client.cli.CommandLineOptions;
+import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.runtime.instance.ActorGateway;
import org.apache.flink.yarn.cli.FlinkYarnSessionCli;
import org.apache.flink.test.util.TestBaseUtils;
import org.apache.flink.util.TestLogger;
@@ -43,10 +47,14 @@ import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.rules.TemporaryFolder;
+import org.mockito.Mockito;
+import org.mockito.verification.VerificationMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.Marker;
import org.slf4j.MarkerFactory;
+import scala.concurrent.Await;
+import scala.concurrent.duration.FiniteDuration;
import java.io.ByteArrayOutputStream;
import java.io.File;
@@ -61,6 +69,7 @@ import java.util.List;
import java.util.Map;
import java.util.Scanner;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;
@@ -562,28 +571,49 @@ public abstract class YarnTestBase extends TestLogger {
@Override
public void run() {
- switch(type) {
- case YARN_SESSION:
- yCli = new FlinkYarnSessionCli("", "", false);
- returnValue = yCli.run(args);
- break;
- case CLI_FRONTEND:
- try {
- CliFrontend cli = new CliFrontend();
- returnValue = cli.parseParameters(args);
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- break;
- default:
- throw new RuntimeException("Unknown type " + type);
- }
+ try {
+ switch (type) {
+ case YARN_SESSION:
+ yCli = new FlinkYarnSessionCli("", "", false);
+ returnValue = yCli.run(args);
+ break;
+ case CLI_FRONTEND:
+ TestingCLI cli;
+ try {
+ cli = new TestingCLI();
+ returnValue = cli.parseParameters(args);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to execute the following args with CliFrontend: "
+ + Arrays.toString(args), e);
+ }
- if(returnValue != 0) {
- Assert.fail("The YARN session returned with non-null value="+returnValue);
+ final ClusterClient client = cli.getClusterClient();
+ try {
+ // check if the JobManager is still alive after running the job
+ final FiniteDuration finiteDuration = new FiniteDuration(10, TimeUnit.SECONDS);
+ ActorGateway jobManagerGateway = client.getJobManagerGateway();
+ Await.ready(jobManagerGateway.ask(new Identify(true), finiteDuration), finiteDuration);
+ } catch (Exception e) {
+ throw new RuntimeException("It seems like the JobManager died although it should still be alive");
+ }
+ // verify we would have shut down anyways and then shutdown
+ Mockito.verify(cli.getSpiedClusterClient()).shutdown();
+ client.shutdown();
+
+ break;
+ default:
+ throw new RuntimeException("Unknown type " + type);
+ }
+
+ if (returnValue != 0) {
+ Assert.fail("The YARN session returned with non-null value=" + returnValue);
+ }
+ } catch (Throwable t) {
+ Assert.fail(t.getMessage());
}
}
+ /** Stops the Yarn session */
public void sendStop() {
if(yCli != null) {
yCli.stop();
@@ -623,4 +653,30 @@ public abstract class YarnTestBase extends TestLogger {
return System.getenv("TRAVIS") != null && System.getenv("TRAVIS").equals("true");
}
+ private static class TestingCLI extends CliFrontend {
+
+ private ClusterClient originalClusterClient;
+ private ClusterClient spiedClusterClient;
+
+ public TestingCLI() throws Exception {}
+
+ @Override
+ protected ClusterClient createClient(CommandLineOptions options, String programName) throws Exception {
+ // mock the returned ClusterClient to disable shutdown and verify shutdown behavior later on
+ originalClusterClient = super.createClient(options, programName);
+ spiedClusterClient = Mockito.spy(originalClusterClient);
+ Mockito.doNothing().when(spiedClusterClient).shutdown();
+ return spiedClusterClient;
+ }
+
+ public ClusterClient getClusterClient() {
+ return originalClusterClient;
+ }
+
+ public ClusterClient getSpiedClusterClient() {
+ return spiedClusterClient;
+ }
+
+
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/28da9954/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java
index dfc71e0..e76b7e8 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java
@@ -198,11 +198,10 @@ public class YarnClusterClient extends ClusterClient {
@Override
protected JobSubmissionResult submitJob(JobGraph jobGraph, ClassLoader classLoader) throws ProgramInvocationException {
- if (perJobCluster) {
- stopAfterJob(jobGraph.getJobID());
- }
-
if (isDetached()) {
+ if (perJobCluster) {
+ stopAfterJob(jobGraph.getJobID());
+ }
return super.runDetached(jobGraph, classLoader);
} else {
return super.run(jobGraph, classLoader);
@@ -248,7 +247,7 @@ public class YarnClusterClient extends ClusterClient {
throw new RuntimeException("Unable to get ClusterClient status from Application Client", e);
}
if(clusterStatus instanceof None$) {
- return null;
+ throw new RuntimeException("Unable to get ClusterClient status from Application Client");
} else if(clusterStatus instanceof Some) {
return (GetClusterStatusResponse) (((Some) clusterStatus).get());
} else {
@@ -572,7 +571,6 @@ public class YarnClusterClient extends ClusterClient {
Thread.sleep(250);
} catch (InterruptedException e) {
LOG.error("Interrupted while waiting for TaskManagers");
- System.err.println("Thread is interrupted");
throw new RuntimeException("Interrupted while waiting for TaskManagers", e);
}
}
[2/3] flink git commit: [FLINK-4486] detached YarnSession: wait until
cluster startup is complete
Posted by mx...@apache.org.
[FLINK-4486] detached YarnSession: wait until cluster startup is complete
This closes #2423
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/4cdeb118
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/4cdeb118
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/4cdeb118
Branch: refs/heads/release-1.1
Commit: 4cdeb11854956ac6cf1189d7cfa43628fb3be328
Parents: 28da995
Author: Maximilian Michels <mx...@apache.org>
Authored: Fri Aug 19 14:57:14 2016 +0200
Committer: Maximilian Michels <mx...@apache.org>
Committed: Mon Aug 29 18:20:00 2016 +0200
----------------------------------------------------------------------
.../flink/client/program/ClusterClient.java | 17 ++-
.../client/program/StandaloneClusterClient.java | 3 +
.../apache/flink/yarn/YarnClusterClient.java | 112 ++++++++++---------
.../flink/yarn/cli/FlinkYarnSessionCli.java | 1 +
4 files changed, 79 insertions(+), 54 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/4cdeb118/flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java b/flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java
index c3c666b..b8d3400 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java
@@ -386,7 +386,10 @@ public abstract class ClusterClient {
* @throws ProgramInvocationException
*/
public JobExecutionResult run(JobGraph jobGraph, ClassLoader classLoader) throws ProgramInvocationException {
- LeaderRetrievalService leaderRetrievalService;
+
+ waitForClusterToBeReady();
+
+ final LeaderRetrievalService leaderRetrievalService;
try {
leaderRetrievalService = LeaderRetrievalUtils.createLeaderRetrievalService(flinkConfig);
} catch (Exception e) {
@@ -411,8 +414,10 @@ public abstract class ClusterClient {
* @throws ProgramInvocationException
*/
public JobSubmissionResult runDetached(JobGraph jobGraph, ClassLoader classLoader) throws ProgramInvocationException {
- ActorGateway jobManagerGateway;
+ waitForClusterToBeReady();
+
+ final ActorGateway jobManagerGateway;
try {
jobManagerGateway = getJobManagerGateway();
} catch (Exception e) {
@@ -655,6 +660,14 @@ public abstract class ClusterClient {
// ------------------------------------------------------------------------
/**
+ * Blocks until the client has determined that the cluster is ready for Job submission.
+ *
+ * This is delayed until right before job submission to report any other errors first
+ * (e.g. invalid job definitions/errors in the user jar)
+ */
+ public abstract void waitForClusterToBeReady();
+
+ /**
* Returns an URL (as a string) to the JobManager web interface
*/
public abstract String getWebInterfaceURL();
http://git-wip-us.apache.org/repos/asf/flink/blob/4cdeb118/flink-clients/src/main/java/org/apache/flink/client/program/StandaloneClusterClient.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/StandaloneClusterClient.java b/flink-clients/src/main/java/org/apache/flink/client/program/StandaloneClusterClient.java
index d25c9d1..3343b69 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/StandaloneClusterClient.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/StandaloneClusterClient.java
@@ -41,6 +41,9 @@ public class StandaloneClusterClient extends ClusterClient {
super(config);
}
+ @Override
+ public void waitForClusterToBeReady() {}
+
@Override
public String getWebInterfaceURL() {
http://git-wip-us.apache.org/repos/asf/flink/blob/4cdeb118/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java
index e76b7e8..75bfeed 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java
@@ -90,7 +90,8 @@ public class YarnClusterClient extends ClusterClient {
private boolean isConnected = true;
- private final boolean perJobCluster;
+ /** Indicator whether this cluster has just been created */
+ private final boolean newlyCreatedCluster;
/**
* Create a new Flink on YARN cluster.
@@ -100,7 +101,7 @@ public class YarnClusterClient extends ClusterClient {
* @param appReport the YARN application ID
* @param flinkConfig Flink configuration
* @param sessionFilesDir Location of files required for YARN session
- * @param perJobCluster Indicator whether this cluster is only created for a single job and then shutdown
+ * @param newlyCreatedCluster Indicator whether this cluster has just been created
* @throws IOException
* @throws YarnException
*/
@@ -110,7 +111,7 @@ public class YarnClusterClient extends ClusterClient {
final ApplicationReport appReport,
org.apache.flink.configuration.Configuration flinkConfig,
Path sessionFilesDir,
- boolean perJobCluster) throws IOException, YarnException {
+ boolean newlyCreatedCluster) throws IOException, YarnException {
super(flinkConfig);
@@ -123,13 +124,13 @@ public class YarnClusterClient extends ClusterClient {
this.appReport = appReport;
this.appId = appReport.getApplicationId();
this.trackingURL = appReport.getTrackingUrl();
- this.perJobCluster = perJobCluster;
+ this.newlyCreatedCluster = newlyCreatedCluster;
- this.applicationClient = new LazApplicationClientLoader();
+ this.applicationClient = new LazApplicationClientLoader(flinkConfig, actorSystemLoader);
- pollingRunner = new PollingThread(yarnClient, appId);
- pollingRunner.setDaemon(true);
- pollingRunner.start();
+ this.pollingRunner = new PollingThread(yarnClient, appId);
+ this.pollingRunner.setDaemon(true);
+ this.pollingRunner.start();
Runtime.getRuntime().addShutdownHook(clientShutdownHook);
}
@@ -199,7 +200,7 @@ public class YarnClusterClient extends ClusterClient {
@Override
protected JobSubmissionResult submitJob(JobGraph jobGraph, ClassLoader classLoader) throws ProgramInvocationException {
if (isDetached()) {
- if (perJobCluster) {
+ if (newlyCreatedCluster) {
stopAfterJob(jobGraph.getJobID());
}
return super.runDetached(jobGraph, classLoader);
@@ -342,8 +343,7 @@ public class YarnClusterClient extends ClusterClient {
*/
@Override
public void finalizeCluster() {
- if (isDetached() || !perJobCluster) {
- // only disconnect if we are not running a per job cluster
+ if (isDetached() || !newlyCreatedCluster) {
disconnect();
} else {
shutdownCluster();
@@ -369,20 +369,16 @@ public class YarnClusterClient extends ClusterClient {
// we are already in the shutdown hook
}
- if(actorSystemLoader.isLoaded()){
- LOG.info("Sending shutdown request to the Application Master");
- if(applicationClient.get() != ActorRef.noSender()) {
- try {
- Future<Object> response =
- Patterns.ask(applicationClient.get(),
- new YarnMessages.LocalStopYarnSession(getApplicationStatus(),
- "Flink YARN Client requested shutdown"),
- new Timeout(akkaDuration));
- Await.ready(response, akkaDuration);
- } catch(Exception e) {
- LOG.warn("Error while stopping YARN Application Client", e);
- }
- }
+ LOG.info("Sending shutdown request to the Application Master");
+ try {
+ Future<Object> response =
+ Patterns.ask(applicationClient.get(),
+ new YarnMessages.LocalStopYarnSession(getApplicationStatus(),
+ "Flink YARN Client requested shutdown"),
+ new Timeout(akkaDuration));
+ Await.ready(response, akkaDuration);
+ } catch(Exception e) {
+ LOG.warn("Error while stopping YARN cluster.", e);
}
try {
@@ -518,14 +514,52 @@ public class YarnClusterClient extends ClusterClient {
return super.isDetached() || clusterDescriptor.isDetachedMode();
}
+ /**
+ * Blocks until all TaskManagers are connected to the JobManager.
+ */
+ @Override
+ public void waitForClusterToBeReady() {
+ logAndSysout("Waiting until all TaskManagers have connected");
+
+ for (GetClusterStatusResponse currentStatus, lastStatus = null;; lastStatus = currentStatus) {
+ currentStatus = getClusterStatus();
+ if (currentStatus != null && !currentStatus.equals(lastStatus)) {
+ logAndSysout("TaskManager status (" + currentStatus.numRegisteredTaskManagers() + "/"
+ + clusterDescriptor.getTaskManagerCount() + ")");
+ if (currentStatus.numRegisteredTaskManagers() >= clusterDescriptor.getTaskManagerCount()) {
+ logAndSysout("All TaskManagers are connected");
+ break;
+ }
+ } else if (lastStatus == null) {
+ logAndSysout("No status updates from the YARN cluster received so far. Waiting ...");
+ }
+
+ try {
+ Thread.sleep(250);
+ } catch (InterruptedException e) {
+ throw new RuntimeException("Interrupted while waiting for TaskManagers", e);
+ }
+ }
+ }
+
public ApplicationId getApplicationId() {
return appId;
}
- protected class LazApplicationClientLoader {
+ private static class LazApplicationClientLoader {
+
+ private final org.apache.flink.configuration.Configuration flinkConfig;
+ private final LazyActorSystemLoader actorSystemLoader;
private ActorRef applicationClient;
+ private LazApplicationClientLoader(
+ org.apache.flink.configuration.Configuration flinkConfig,
+ LazyActorSystemLoader actorSystemLoader) {
+ this.flinkConfig = flinkConfig;
+ this.actorSystemLoader = actorSystemLoader;
+ }
+
/**
* Creates a new ApplicationClient actor or returns an existing one. May start an ActorSystem.
* @return ActorSystem
@@ -549,32 +583,6 @@ public class YarnClusterClient extends ClusterClient {
flinkConfig,
leaderRetrievalService),
"applicationClient");
-
- if (perJobCluster) {
-
- logAndSysout("Waiting until all TaskManagers have connected");
-
- for (GetClusterStatusResponse currentStatus, lastStatus = null;; lastStatus = currentStatus) {
- currentStatus = getClusterStatus();
- if (currentStatus != null && !currentStatus.equals(lastStatus)) {
- logAndSysout("TaskManager status (" + currentStatus.numRegisteredTaskManagers() + "/"
- + clusterDescriptor.getTaskManagerCount() + ")");
- if (currentStatus.numRegisteredTaskManagers() >= clusterDescriptor.getTaskManagerCount()) {
- logAndSysout("All TaskManagers are connected");
- break;
- }
- } else if (lastStatus == null) {
- logAndSysout("No status updates from the YARN cluster received so far. Waiting ...");
- }
-
- try {
- Thread.sleep(250);
- } catch (InterruptedException e) {
- LOG.error("Interrupted while waiting for TaskManagers");
- throw new RuntimeException("Interrupted while waiting for TaskManagers", e);
- }
- }
- }
}
return applicationClient;
http://git-wip-us.apache.org/repos/asf/flink/blob/4cdeb118/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java b/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
index bee6a7a..28d8fb8 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
@@ -628,6 +628,7 @@ public class FlinkYarnSessionCli implements CustomCommandLine<YarnClusterClient>
"yarn application -kill " + yarnCluster.getApplicationId() + System.lineSeparator() +
"Please also note that the temporary files of the YARN session in {} will not be removed.",
yarnDescriptor.getSessionFilesDir());
+ yarnCluster.waitForClusterToBeReady();
yarnCluster.disconnect();
} else {
runInteractiveCli(yarnCluster, acceptInteractiveInput);
[3/3] flink git commit: [FLINK-4526][yarn] remove redundant proxy
messages
Posted by mx...@apache.org.
[FLINK-4526][yarn] remove redundant proxy messages
This closes #2437
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/f7ec1efb
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/f7ec1efb
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/f7ec1efb
Branch: refs/heads/release-1.1
Commit: f7ec1efb32c277a0d51479cca4450edb324a3d58
Parents: 4cdeb11
Author: Maximilian Michels <mx...@apache.org>
Authored: Tue Aug 30 10:46:22 2016 +0200
Committer: Maximilian Michels <mx...@apache.org>
Committed: Mon Aug 29 18:20:30 2016 +0200
----------------------------------------------------------------------
.../flink/client/program/ClusterClient.java | 3 +-
.../apache/flink/yarn/YarnClusterClient.java | 41 +++++++-------------
.../apache/flink/yarn/ApplicationClient.scala | 27 -------------
.../org/apache/flink/yarn/YarnMessages.scala | 7 ----
4 files changed, 15 insertions(+), 63 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/f7ec1efb/flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java b/flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java
index b8d3400..2a0aca3 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java
@@ -636,8 +636,7 @@ public abstract class ClusterClient {
* @throws Exception
*/
public ActorGateway getJobManagerGateway() throws Exception {
- LOG.info("Looking up JobManager");
-
+ LOG.debug("Looking up JobManager");
return LeaderRetrievalUtils.retrieveLeaderGateway(
LeaderRetrievalUtils.createLeaderRetrievalService(flinkConfig),
actorSystemLoader.get(),
http://git-wip-us.apache.org/repos/asf/flink/blob/f7ec1efb/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java
index 75bfeed..8b6cd9a 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java
@@ -19,8 +19,6 @@ package org.apache.flink.yarn;
import akka.actor.ActorRef;
-import static akka.pattern.Patterns.ask;
-
import akka.actor.Props;
import akka.pattern.Patterns;
import akka.util.Timeout;
@@ -30,8 +28,10 @@ import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.client.program.ProgramInvocationException;
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.clusterframework.ApplicationStatus;
+import org.apache.flink.runtime.clusterframework.messages.GetClusterStatus;
import org.apache.flink.runtime.clusterframework.messages.GetClusterStatusResponse;
import org.apache.flink.runtime.clusterframework.messages.InfoMessage;
+import org.apache.flink.runtime.clusterframework.messages.ShutdownClusterAfterJob;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
import org.apache.flink.runtime.util.LeaderRetrievalUtils;
@@ -48,9 +48,7 @@ import org.apache.hadoop.yarn.client.api.YarnClient;
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import scala.None$;
import scala.Option;
-import scala.Some;
import scala.concurrent.Await;
import scala.concurrent.Future;
import scala.concurrent.duration.FiniteDuration;
@@ -83,7 +81,6 @@ public class YarnClusterClient extends ClusterClient {
private final AbstractYarnClusterDescriptor clusterDescriptor;
private final LazApplicationClientLoader applicationClient;
private final FiniteDuration akkaDuration;
- private final Timeout akkaTimeout;
private final ApplicationReport appReport;
private final ApplicationId appId;
private final String trackingURL;
@@ -116,7 +113,6 @@ public class YarnClusterClient extends ClusterClient {
super(flinkConfig);
this.akkaDuration = AkkaUtils.getTimeout(flinkConfig);
- this.akkaTimeout = Timeout.durationToTimeout(akkaDuration);
this.clusterDescriptor = clusterDescriptor;
this.yarnClient = yarnClient;
this.hadoopConfig = yarnClient.getConfig();
@@ -175,12 +171,12 @@ public class YarnClusterClient extends ClusterClient {
*/
private void stopAfterJob(JobID jobID) {
Preconditions.checkNotNull(jobID, "The job id must not be null");
- Future<Object> messageReceived =
- ask(
- applicationClient.get(),
- new YarnMessages.LocalStopAMAfterJob(jobID), akkaTimeout);
try {
- Await.result(messageReceived, akkaDuration);
+ Future<Object> replyFuture =
+ getJobManagerGateway().ask(
+ new ShutdownClusterAfterJob(jobID),
+ akkaDuration);
+ Await.ready(replyFuture, akkaDuration);
} catch (Exception e) {
throw new RuntimeException("Unable to tell application master to stop once the specified job has been finised", e);
}
@@ -230,30 +226,21 @@ public class YarnClusterClient extends ClusterClient {
@Override
public GetClusterStatusResponse getClusterStatus() {
if(!isConnected) {
- throw new IllegalStateException("The cluster is not connected to the ApplicationMaster.");
+ throw new IllegalStateException("The cluster is not connected to the cluster.");
}
if(hasBeenShutdown()) {
- return null;
+ throw new IllegalStateException("The cluster has already been shutdown.");
}
- Future<Object> clusterStatusOption =
- ask(
- applicationClient.get(),
- YarnMessages.getLocalGetyarnClusterStatus(),
- akkaTimeout);
- Object clusterStatus;
try {
- clusterStatus = Await.result(clusterStatusOption, akkaDuration);
+ final Future<Object> clusterStatusOption =
+ getJobManagerGateway().ask(
+ GetClusterStatus.getInstance(),
+ akkaDuration);
+ return (GetClusterStatusResponse) Await.result(clusterStatusOption, akkaDuration);
} catch (Exception e) {
throw new RuntimeException("Unable to get ClusterClient status from Application Client", e);
}
- if(clusterStatus instanceof None$) {
- throw new RuntimeException("Unable to get ClusterClient status from Application Client");
- } else if(clusterStatus instanceof Some) {
- return (GetClusterStatusResponse) (((Some) clusterStatus).get());
- } else {
- throw new RuntimeException("Unexpected type: " + clusterStatus.getClass().getCanonicalName());
- }
}
public ApplicationStatus getApplicationStatus() {
http://git-wip-us.apache.org/repos/asf/flink/blob/f7ec1efb/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationClient.scala
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationClient.scala b/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationClient.scala
index e701269..7442503 100644
--- a/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationClient.scala
+++ b/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationClient.scala
@@ -48,7 +48,6 @@ class ApplicationClient(
with LeaderSessionMessageFilter
with LogMessages
with LeaderRetrievalListener{
- import context._
val log = Logger(getClass)
@@ -60,7 +59,6 @@ class ApplicationClient(
var pollingTimer: Option[Cancellable] = None
var running = false
var messagesQueue : mutable.Queue[InfoMessage] = mutable.Queue[InfoMessage]()
- var latestClusterStatus : Option[GetClusterStatusResponse] = None
var stopMessageReceiver : Option[ActorRef] = None
var leaderSessionID: Option[UUID] = None
@@ -136,19 +134,8 @@ class ApplicationClient(
// The job manager acts as a proxy between the client and the resource managert
val jm = sender()
log.info(s"Successfully registered at the ResourceManager using JobManager $jm")
-
yarnJobManager = Some(jm)
- // schedule a periodic status report from the JobManager
- // request the number of task managers and slots from the job manager
- pollingTimer = Some(
- context.system.scheduler.schedule(
- INITIAL_POLLING_DELAY,
- WAIT_FOR_YARN_INTERVAL,
- yarnJobManager.get,
- decorateMessage(GetClusterStatus.getInstance()))
- )
-
case JobManagerLeaderAddress(jobManagerAkkaURL, newLeaderSessionID) =>
log.info(s"Received address of new leader $jobManagerAkkaURL with session ID" +
s" $newLeaderSessionID.")
@@ -192,20 +179,6 @@ class ApplicationClient(
}(context.dispatcher)
}
- // handle the responses from the PollYarnClusterStatus messages to the yarn job mgr
- case status: GetClusterStatusResponse =>
- latestClusterStatus = Some(status)
-
- // locally get cluster status
- case LocalGetYarnClusterStatus =>
- sender() ! decorateMessage(latestClusterStatus)
-
- // Forward message to Application Master
- case LocalStopAMAfterJob(jobID) =>
- yarnJobManager foreach {
- _ forward decorateMessage(new ShutdownClusterAfterJob(jobID))
- }
-
// ----------------- handle messages from the cluster -------------------
// receive remote messages
case msg: InfoMessage =>
http://git-wip-us.apache.org/repos/asf/flink/blob/f7ec1efb/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnMessages.scala
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnMessages.scala b/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnMessages.scala
index da1917b..ada2631 100644
--- a/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnMessages.scala
+++ b/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnMessages.scala
@@ -86,17 +86,10 @@ object YarnMessages {
case object HeartbeatWithYarn
case object CheckForUserCommand
- // tell the AM to monitor the job and stop once it has finished
- case class LocalStopAMAfterJob(jobId:JobID)
-
case object LocalGetYarnMessage // request new message
- case object LocalGetYarnClusterStatus // request the latest cluster status
def getLocalGetYarnMessage(): AnyRef = {
LocalGetYarnMessage
}
- def getLocalGetyarnClusterStatus(): AnyRef = {
- LocalGetYarnClusterStatus
- }
}