You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@twill.apache.org by ch...@apache.org on 2014/03/12 21:08:22 UTC
git commit: (TWILL-52) Improve ZK session expire handling. Avoid
killing application if expiration happened by resorting to polling YARN for
app status, until the AM is reconnected to ZK.
Repository: incubator-twill
Updated Branches:
refs/heads/master c05a48d0f -> 3d3c0e92e
(TWILL-52) Improve ZK session expire handling. Avoid killing application if expiration happened by resorting to polling YARN for app status, until the AM is reconnected to ZK.
Signed-off-by: Terence Yim <te...@continuuity.com>
Project: http://git-wip-us.apache.org/repos/asf/incubator-twill/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-twill/commit/3d3c0e92
Tree: http://git-wip-us.apache.org/repos/asf/incubator-twill/tree/3d3c0e92
Diff: http://git-wip-us.apache.org/repos/asf/incubator-twill/diff/3d3c0e92
Branch: refs/heads/master
Commit: 3d3c0e92e8b7455b247202763fc7a2c8a2ff5b71
Parents: c05a48d
Author: Terence Yim <ch...@apache.org>
Authored: Thu Mar 6 02:16:10 2014 -0800
Committer: Terence Yim <te...@continuuity.com>
Committed: Wed Mar 12 13:02:19 2014 -0700
----------------------------------------------------------------------
.../internal/AbstractZKServiceController.java | 85 +++++++----
.../twill/internal/TwillContainerLauncher.java | 5 +
.../twill/internal/ZKServiceDecorator.java | 28 ++++
.../apache/twill/internal/ControllerTest.java | 9 ++
.../apache/twill/yarn/YarnTwillController.java | 129 ++++++++++++++--
.../twill/yarn/YarnTwillRunnerService.java | 4 +-
.../apache/twill/yarn/SessionExpireTestRun.java | 147 +++++++++++++++++++
.../org/apache/twill/yarn/YarnTestSuite.java | 3 +-
.../org/apache/twill/yarn/YarnTestUtils.java | 4 +
9 files changed, 366 insertions(+), 48 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/3d3c0e92/twill-core/src/main/java/org/apache/twill/internal/AbstractZKServiceController.java
----------------------------------------------------------------------
diff --git a/twill-core/src/main/java/org/apache/twill/internal/AbstractZKServiceController.java b/twill-core/src/main/java/org/apache/twill/internal/AbstractZKServiceController.java
index a132128..80932ea 100644
--- a/twill-core/src/main/java/org/apache/twill/internal/AbstractZKServiceController.java
+++ b/twill-core/src/main/java/org/apache/twill/internal/AbstractZKServiceController.java
@@ -53,7 +53,7 @@ public abstract class AbstractZKServiceController extends AbstractExecutionServi
private static final Logger LOG = LoggerFactory.getLogger(AbstractZKServiceController.class);
- private final ZKClient zkClient;
+ protected final ZKClient zkClient;
private final InstanceNodeDataCallback instanceNodeDataCallback;
private final StateNodeDataCallback stateNodeDataCallback;
private final List<ListenableFuture<?>> messageFutures;
@@ -79,6 +79,8 @@ public abstract class AbstractZKServiceController extends AbstractExecutionServi
@Override
protected final void startUp() {
+ doStartUp();
+
// Watch for instance node existence.
actOnExists(getInstancePath(), new Runnable() {
@Override
@@ -94,8 +96,6 @@ public abstract class AbstractZKServiceController extends AbstractExecutionServi
watchStateNode();
}
});
-
- doStartUp();
}
@Override
@@ -165,6 +165,12 @@ public abstract class AbstractZKServiceController extends AbstractExecutionServi
protected abstract void instanceNodeUpdated(NodeData nodeData);
/**
+ * Called when failed to fetch from live instance node.
+ * @param cause The cause of failure of {@code null} if cause is unknown.
+ */
+ protected abstract void instanceNodeFailed(Throwable cause);
+
+ /**
* Called when an update on the state node is detected.
* @param stateNode The update state node data or {@code null} if there is an error when fetching the node data.
*/
@@ -185,6 +191,9 @@ public abstract class AbstractZKServiceController extends AbstractExecutionServi
Futures.addCallback(zkClient.exists(path, new Watcher() {
@Override
public void process(WatchedEvent event) {
+ if (!shouldProcessZKEvent()) {
+ return;
+ }
// When node is created, call the action.
// Other event type would be handled by the action.
if (event.getType() == Event.EventType.NodeCreated && nodeExists.compareAndSet(false, true)) {
@@ -207,13 +216,14 @@ public abstract class AbstractZKServiceController extends AbstractExecutionServi
}, Threads.SAME_THREAD_EXECUTOR);
}
- private void watchInstanceNode() {
+ protected final void watchInstanceNode() {
+ if (!shouldProcessZKEvent()) {
+ return;
+ }
Futures.addCallback(zkClient.getData(getInstancePath(), new Watcher() {
@Override
public void process(WatchedEvent event) {
- State state = state();
- if (state != State.NEW && state != State.STARTING && state != State.RUNNING) {
- // Ignore ZK node events when it is in stopping sequence.
+ if (!shouldProcessZKEvent()) {
return;
}
switch (event.getType()) {
@@ -221,8 +231,7 @@ public abstract class AbstractZKServiceController extends AbstractExecutionServi
watchInstanceNode();
break;
case NodeDeleted:
- // When the ephemeral node goes away, treat the remote service stopped.
- forceShutDown();
+ instanceNodeFailed(KeeperException.create(KeeperException.Code.NONODE, getInstancePath()));
break;
default:
LOG.info("Ignore ZK event for instance node: {}", event);
@@ -232,12 +241,13 @@ public abstract class AbstractZKServiceController extends AbstractExecutionServi
}
private void watchStateNode() {
+ if (!shouldProcessZKEvent()) {
+ return;
+ }
Futures.addCallback(zkClient.getData(getZKPath("state"), new Watcher() {
@Override
public void process(WatchedEvent event) {
- State state = state();
- if (state != State.NEW && state != State.STARTING && state != State.RUNNING) {
- // Ignore ZK node events when it is in stopping sequence.
+ if (!shouldProcessZKEvent()) {
return;
}
switch (event.getType()) {
@@ -245,13 +255,21 @@ public abstract class AbstractZKServiceController extends AbstractExecutionServi
watchStateNode();
break;
default:
- LOG.info("Ignore ZK event for state node: {}", event);
+ LOG.debug("Ignore ZK event for state node: {}", event);
}
}
}), stateNodeDataCallback, Threads.SAME_THREAD_EXECUTOR);
}
/**
+ * Returns true if ZK events needs to be processed.
+ */
+ private boolean shouldProcessZKEvent() {
+ State state = state();
+ return (state == State.NEW || state == State.STARTING || state == State.RUNNING);
+ }
+
+ /**
* Returns the path prefix for creating sequential message node for the remote service.
*/
private String getMessagePrefix() {
@@ -261,7 +279,7 @@ public abstract class AbstractZKServiceController extends AbstractExecutionServi
/**
* Returns the zookeeper node path for the ephemeral instance node for this runId.
*/
- private String getInstancePath() {
+ protected final String getInstancePath() {
return String.format("/instances/%s", getRunId().getId());
}
@@ -273,17 +291,16 @@ public abstract class AbstractZKServiceController extends AbstractExecutionServi
@Override
public void onSuccess(NodeData result) {
- instanceNodeUpdated(result);
+ if (shouldProcessZKEvent()) {
+ instanceNodeUpdated(result);
+ }
}
@Override
public void onFailure(Throwable t) {
LOG.error("Failed in fetching instance node data.", t);
- if (t instanceof KeeperException && ((KeeperException) t).code() == KeeperException.Code.NONODE) {
- // If the node is gone, treat the remote service stopped.
- forceShutDown();
- } else {
- instanceNodeUpdated(null);
+ if (shouldProcessZKEvent()) {
+ instanceNodeFailed(t);
}
}
}
@@ -292,23 +309,27 @@ public abstract class AbstractZKServiceController extends AbstractExecutionServi
@Override
public void onSuccess(NodeData result) {
- byte[] data = result.getData();
- if (data == null) {
- stateNodeUpdated(null);
- return;
- }
- StateNode stateNode = new GsonBuilder().registerTypeAdapter(StateNode.class, new StateNodeCodec())
- .registerTypeAdapter(StackTraceElement.class, new StackTraceElementCodec())
- .create()
- .fromJson(new String(data, Charsets.UTF_8), StateNode.class);
+ if (shouldProcessZKEvent()) {
+ byte[] data = result.getData();
+ if (data == null) {
+ stateNodeUpdated(null);
+ return;
+ }
+ StateNode stateNode = new GsonBuilder().registerTypeAdapter(StateNode.class, new StateNodeCodec())
+ .registerTypeAdapter(StackTraceElement.class, new StackTraceElementCodec())
+ .create()
+ .fromJson(new String(data, Charsets.UTF_8), StateNode.class);
- stateNodeUpdated(stateNode);
+ stateNodeUpdated(stateNode);
+ }
}
@Override
public void onFailure(Throwable t) {
- LOG.error("Failed in fetching state node data.", t);
- stateNodeUpdated(null);
+ if (shouldProcessZKEvent()) {
+ LOG.error("Failed in fetching state node data.", t);
+ stateNodeUpdated(null);
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/3d3c0e92/twill-core/src/main/java/org/apache/twill/internal/TwillContainerLauncher.java
----------------------------------------------------------------------
diff --git a/twill-core/src/main/java/org/apache/twill/internal/TwillContainerLauncher.java b/twill-core/src/main/java/org/apache/twill/internal/TwillContainerLauncher.java
index 5d2f33c..430e63a 100644
--- a/twill-core/src/main/java/org/apache/twill/internal/TwillContainerLauncher.java
+++ b/twill-core/src/main/java/org/apache/twill/internal/TwillContainerLauncher.java
@@ -163,6 +163,11 @@ public final class TwillContainerLauncher {
}
@Override
+ protected void instanceNodeFailed(Throwable cause) {
+ // No-op
+ }
+
+ @Override
protected void stateNodeUpdated(StateNode stateNode) {
// No-op
}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/3d3c0e92/twill-core/src/main/java/org/apache/twill/internal/ZKServiceDecorator.java
----------------------------------------------------------------------
diff --git a/twill-core/src/main/java/org/apache/twill/internal/ZKServiceDecorator.java b/twill-core/src/main/java/org/apache/twill/internal/ZKServiceDecorator.java
index 7592176..efefe36 100644
--- a/twill-core/src/main/java/org/apache/twill/internal/ZKServiceDecorator.java
+++ b/twill-core/src/main/java/org/apache/twill/internal/ZKServiceDecorator.java
@@ -110,6 +110,7 @@ public final class ZKServiceDecorator extends AbstractService {
@Override
protected void doStart() {
callbackExecutor = Executors.newSingleThreadExecutor(Threads.createDaemonThreadFactory("message-callback"));
+ // Create the live node, if succeeded, start the decorated service, otherwise fail out.
Futures.addCallback(createLiveNode(), new FutureCallback<String>() {
@Override
public void onSuccess(String result) {
@@ -142,6 +143,33 @@ public final class ZKServiceDecorator extends AbstractService {
notifyFailed(t);
}
});
+
+ // Watch for session expiration, recreate the live node if reconnected after expiration.
+ zkClient.addConnectionWatcher(new Watcher() {
+ private boolean expired = false;
+
+ @Override
+ public void process(WatchedEvent event) {
+ if (event.getState() == Event.KeeperState.Expired) {
+ LOG.warn("ZK Session expired for service {} with runId {}.", decoratedService, id.getId());
+ expired = true;
+ } else if (event.getState() == Event.KeeperState.SyncConnected && expired) {
+ LOG.info("Reconnected after expiration for service {} with runId {}", decoratedService, id.getId());
+ expired = false;
+ Futures.addCallback(createLiveNode(), new FutureCallback<String>() {
+ @Override
+ public void onSuccess(String result) {
+ // All good, no-op
+ }
+
+ @Override
+ public void onFailure(Throwable t) {
+ notifyFailed(t);
+ }
+ }, Threads.SAME_THREAD_EXECUTOR);
+ }
+ }
+ });
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/3d3c0e92/twill-core/src/test/java/org/apache/twill/internal/ControllerTest.java
----------------------------------------------------------------------
diff --git a/twill-core/src/test/java/org/apache/twill/internal/ControllerTest.java b/twill-core/src/test/java/org/apache/twill/internal/ControllerTest.java
index 048e489..4d968bc 100644
--- a/twill-core/src/test/java/org/apache/twill/internal/ControllerTest.java
+++ b/twill-core/src/test/java/org/apache/twill/internal/ControllerTest.java
@@ -35,6 +35,7 @@ import org.apache.twill.internal.zookeeper.InMemoryZKServer;
import org.apache.twill.zookeeper.NodeData;
import org.apache.twill.zookeeper.ZKClient;
import org.apache.twill.zookeeper.ZKClientService;
+import org.apache.zookeeper.KeeperException;
import org.junit.Assert;
import org.junit.Test;
import org.slf4j.Logger;
@@ -196,6 +197,14 @@ public class ControllerTest {
}
@Override
+ protected void instanceNodeFailed(Throwable cause) {
+ // Shutdown if the instance node goes away
+ if (cause instanceof KeeperException.NoNodeException) {
+ forceShutDown();
+ }
+ }
+
+ @Override
protected void stateNodeUpdated(StateNode stateNode) {
// No-op
}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/3d3c0e92/twill-yarn/src/main/java/org/apache/twill/yarn/YarnTwillController.java
----------------------------------------------------------------------
diff --git a/twill-yarn/src/main/java/org/apache/twill/yarn/YarnTwillController.java b/twill-yarn/src/main/java/org/apache/twill/yarn/YarnTwillController.java
index 8e0c6be..5afd679 100644
--- a/twill-yarn/src/main/java/org/apache/twill/yarn/YarnTwillController.java
+++ b/twill-yarn/src/main/java/org/apache/twill/yarn/YarnTwillController.java
@@ -22,6 +22,7 @@ import com.google.common.collect.ImmutableList;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.Uninterruptibles;
import org.apache.commons.lang.time.StopWatch;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
import org.apache.hadoop.yarn.api.records.YarnApplicationState;
import org.apache.twill.api.ResourceReport;
@@ -37,6 +38,7 @@ import org.apache.twill.internal.state.SystemMessages;
import org.apache.twill.internal.yarn.YarnApplicationReport;
import org.apache.twill.zookeeper.NodeData;
import org.apache.twill.zookeeper.ZKClient;
+import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -44,7 +46,9 @@ import java.io.IOException;
import java.net.URI;
import java.net.URL;
import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
/**
* A {@link org.apache.twill.api.TwillController} that controllers application running on Hadoop YARN.
@@ -53,20 +57,27 @@ final class YarnTwillController extends AbstractTwillController implements Twill
private static final Logger LOG = LoggerFactory.getLogger(YarnTwillController.class);
+ private final String appName;
private final Callable<ProcessController<YarnApplicationReport>> startUp;
private ProcessController<YarnApplicationReport> processController;
private ResourceReportClient resourcesClient;
+ // Thread for polling yarn for application status if application got ZK session expire.
+ // Only used by the instanceUpdate/Delete method, which is from serialized call from ZK callback.
+ private Thread statusPollingThread;
+
/**
* Creates an instance without any {@link LogHandler}.
*/
- YarnTwillController(RunId runId, ZKClient zkClient, Callable<ProcessController<YarnApplicationReport>> startUp) {
- this(runId, zkClient, ImmutableList.<LogHandler>of(), startUp);
+ YarnTwillController(String appName, RunId runId, ZKClient zkClient,
+ Callable<ProcessController<YarnApplicationReport>> startUp) {
+ this(appName, runId, zkClient, ImmutableList.<LogHandler>of(), startUp);
}
- YarnTwillController(RunId runId, ZKClient zkClient, Iterable<LogHandler> logHandlers,
+ YarnTwillController(String appName, RunId runId, ZKClient zkClient, Iterable<LogHandler> logHandlers,
Callable<ProcessController<YarnApplicationReport>> startUp) {
super(runId, zkClient, logHandlers);
+ this.appName = appName;
this.startUp = startUp;
}
@@ -87,7 +98,8 @@ final class YarnTwillController extends AbstractTwillController implements Twill
processController = startUp.call();
YarnApplicationReport report = processController.getReport();
- LOG.debug("Application {} submit", report.getApplicationId());
+ ApplicationId appId = report.getApplicationId();
+ LOG.debug("Application {} with id {} submitted", appName, appId);
YarnApplicationState state = report.getYarnApplicationState();
StopWatch stopWatch = new StopWatch();
@@ -95,18 +107,18 @@ final class YarnTwillController extends AbstractTwillController implements Twill
stopWatch.split();
long maxTime = TimeUnit.MILLISECONDS.convert(Constants.APPLICATION_MAX_START_SECONDS, TimeUnit.SECONDS);
- LOG.info("Checking yarn application status");
+ LOG.debug("Checking yarn application status for {} {}", appName, appId);
while (!hasRun(state) && stopWatch.getSplitTime() < maxTime) {
report = processController.getReport();
state = report.getYarnApplicationState();
- LOG.debug("Yarn application status: {}", state);
+ LOG.debug("Yarn application status for {} {}: {}", appName, appId, state);
TimeUnit.SECONDS.sleep(1);
stopWatch.split();
}
- LOG.info("Yarn application is in state {}", state);
+ LOG.info("Yarn application {} {} is in state {}", appName, appId, state);
if (state != YarnApplicationState.RUNNING) {
- LOG.info("Yarn application is not in running state. Shutting down controller.",
- Constants.APPLICATION_MAX_START_SECONDS);
+ LOG.info("Yarn application {} {} is not in running state. Shutting down controller.",
+ appName, appId, Constants.APPLICATION_MAX_START_SECONDS);
forceShutDown();
} else {
try {
@@ -123,12 +135,15 @@ final class YarnTwillController extends AbstractTwillController implements Twill
}
@Override
- protected void doShutDown() {
+ protected synchronized void doShutDown() {
if (processController == null) {
LOG.warn("No process controller for application that is not submitted.");
return;
}
+ // Stop polling if it is running.
+ stopPollStatus();
+
// Wait for the stop message being processed
try {
Uninterruptibles.getUninterruptibly(getStopMessageFuture(),
@@ -148,13 +163,14 @@ final class YarnTwillController extends AbstractTwillController implements Twill
YarnApplicationReport report = processController.getReport();
FinalApplicationStatus finalStatus = report.getFinalApplicationStatus();
+ ApplicationId appId = report.getApplicationId();
while (finalStatus == FinalApplicationStatus.UNDEFINED && stopWatch.getSplitTime() < maxTime) {
- LOG.debug("Yarn application final status for {} {}", report.getApplicationId(), finalStatus);
+ LOG.debug("Yarn application final status for {} {}: {}", appName, appId, finalStatus);
TimeUnit.SECONDS.sleep(1);
stopWatch.split();
finalStatus = processController.getReport().getFinalApplicationStatus();
}
- LOG.debug("Yarn application final status is {}", finalStatus);
+ LOG.debug("Yarn application {} {} completed with status {}", appName, appId, finalStatus);
// Application not finished after max stop time, kill the application
if (finalStatus == FinalApplicationStatus.UNDEFINED) {
@@ -172,7 +188,7 @@ final class YarnTwillController extends AbstractTwillController implements Twill
public void kill() {
if (processController != null) {
YarnApplicationReport report = processController.getReport();
- LOG.info("Killing application {}", report.getApplicationId());
+ LOG.info("Killing application {} {}", appName, report.getApplicationId());
processController.cancel();
} else {
LOG.warn("No process controller for application that is not submitted.");
@@ -181,7 +197,22 @@ final class YarnTwillController extends AbstractTwillController implements Twill
@Override
protected void instanceNodeUpdated(NodeData nodeData) {
+ }
+
+ @Override
+ protected void instanceNodeFailed(Throwable cause) {
+ // Resort to polling from Yarn for the application status.
+ if (processController == null) {
+ LOG.warn("No process controller for application that is not submitted.");
+ return;
+ }
+ YarnApplicationReport report = processController.getReport();
+
+ // It happens if the application has ZK session expire or the node is deleted due to application termination.
+ LOG.info("Failed to access application {} {} live node in ZK, resort to polling. Failure reason: {}",
+ appName, report.getApplicationId(), cause == null ? "Unknown" : cause.getMessage());
+ startPollStatus(report.getApplicationId());
}
@Override
@@ -189,6 +220,78 @@ final class YarnTwillController extends AbstractTwillController implements Twill
}
+ private synchronized void startPollStatus(ApplicationId appId) {
+ if (statusPollingThread == null) {
+ statusPollingThread = new Thread(createStatusPollingRunnable(),
+ String.format("%s-%s-yarn-poller", appName, appId));
+ statusPollingThread.setDaemon(true);
+ statusPollingThread.start();
+ }
+ }
+
+ private synchronized void stopPollStatus() {
+ if (statusPollingThread != null) {
+ statusPollingThread.interrupt();
+ statusPollingThread = null;
+ }
+ }
+
+ private Runnable createStatusPollingRunnable() {
+ return new Runnable() {
+
+ @Override
+ public void run() {
+ YarnApplicationReport report = processController.getReport();
+ ApplicationId appId = report.getApplicationId();
+ boolean shutdown = false;
+ boolean watchInstanceNode = false;
+
+ try {
+ LOG.debug("Polling status from Yarn for {} {}.", appName, appId);
+ while (!Thread.currentThread().isInterrupted()) {
+ if (report.getFinalApplicationStatus() != FinalApplicationStatus.UNDEFINED) {
+ shutdown = true;
+ break;
+ }
+ // Make a sync exists call to instance node and re-watch if the node exists
+ try {
+ // The timeout is arbitrary, as it's just for avoiding block forever
+ Stat stat = zkClient.exists(getInstancePath()).get(5, TimeUnit.SECONDS);
+ if (stat != null) {
+ watchInstanceNode = true;
+ break;
+ }
+ } catch (ExecutionException e) {
+ // Ignore the exception, as any exception won't affect the status polling.
+ LOG.debug("Failed in exists call on ZK path {}.", getInstancePath(), e);
+ } catch (TimeoutException e) {
+ LOG.debug("Timeout in exists call on ZK path {}.", getInstancePath(), e);
+ }
+
+ TimeUnit.SECONDS.sleep(1);
+ report = processController.getReport();
+ }
+ } catch (InterruptedException e) {
+ // OK to ignore.
+ LOG.debug("Status polling thread interrupted for application {} {}", appName, appId);
+ }
+
+ LOG.debug("Stop polling status from Yarn for {} {}.", appName, appId);
+
+ if (shutdown) {
+ LOG.info("Yarn application {} {} completed. Shutting down controller.", appName, appId);
+ forceShutDown();
+ } else if (watchInstanceNode) {
+ LOG.info("Rewatch instance node for {} {} at {}", appName, appId, getInstancePath());
+ synchronized (YarnTwillController.this) {
+ statusPollingThread = null;
+ watchInstanceNode();
+ }
+ }
+ }
+ };
+ }
+
private boolean hasRun(YarnApplicationState state) {
switch (state) {
case RUNNING:
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/3d3c0e92/twill-yarn/src/main/java/org/apache/twill/yarn/YarnTwillRunnerService.java
----------------------------------------------------------------------
diff --git a/twill-yarn/src/main/java/org/apache/twill/yarn/YarnTwillRunnerService.java b/twill-yarn/src/main/java/org/apache/twill/yarn/YarnTwillRunnerService.java
index 9dff24d..72ea58b 100644
--- a/twill-yarn/src/main/java/org/apache/twill/yarn/YarnTwillRunnerService.java
+++ b/twill-yarn/src/main/java/org/apache/twill/yarn/YarnTwillRunnerService.java
@@ -245,7 +245,7 @@ public final class YarnTwillRunnerService extends AbstractIdleService implements
public YarnTwillController create(RunId runId, Iterable<LogHandler> logHandlers,
Callable<ProcessController<YarnApplicationReport>> startUp) {
ZKClient zkClient = ZKClients.namespace(zkClientService, "/" + appName);
- YarnTwillController controller = listenController(new YarnTwillController(runId, zkClient,
+ YarnTwillController controller = listenController(new YarnTwillController(appName, runId, zkClient,
logHandlers, startUp));
synchronized (YarnTwillRunnerService.this) {
Preconditions.checkArgument(!controllers.contains(appName, runId),
@@ -468,7 +468,7 @@ public final class YarnTwillRunnerService extends AbstractIdleService implements
if (!controllers.contains(appName, runId)) {
ZKClient zkClient = ZKClients.namespace(zkClientService, "/" + appName);
YarnTwillController controller = listenController(
- new YarnTwillController(runId, zkClient,
+ new YarnTwillController(appName, runId, zkClient,
Callables.returning(yarnAppClient.createProcessController(appId))));
controllers.put(appName, runId, controller);
controller.start();
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/3d3c0e92/twill-yarn/src/test/java/org/apache/twill/yarn/SessionExpireTestRun.java
----------------------------------------------------------------------
diff --git a/twill-yarn/src/test/java/org/apache/twill/yarn/SessionExpireTestRun.java b/twill-yarn/src/test/java/org/apache/twill/yarn/SessionExpireTestRun.java
new file mode 100644
index 0000000..ec29061
--- /dev/null
+++ b/twill-yarn/src/test/java/org/apache/twill/yarn/SessionExpireTestRun.java
@@ -0,0 +1,147 @@
+/*
+ * 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.twill.yarn;
+
+import com.google.common.base.Stopwatch;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.Service;
+import org.apache.twill.api.AbstractTwillRunnable;
+import org.apache.twill.api.TwillController;
+import org.apache.twill.api.TwillRunner;
+import org.apache.twill.api.logging.PrinterLogHandler;
+import org.apache.twill.common.ServiceListenerAdapter;
+import org.apache.twill.common.Services;
+import org.apache.twill.common.Threads;
+import org.apache.zookeeper.jmx.MBeanRegistry;
+import org.apache.zookeeper.server.ConnectionMXBean;
+import org.junit.Assert;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.PrintWriter;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import javax.management.MBeanServer;
+import javax.management.MBeanServerInvocationHandler;
+import javax.management.ObjectName;
+import javax.management.Query;
+import javax.management.QueryExp;
+import javax.management.StringValueExp;
+
+/**
+ * Test for testing ZK session expire from AM container
+ */
+public class SessionExpireTestRun extends BaseYarnTest {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SessionExpireTestRun.class);
+
+ @Test
+ public void testAppSessionExpire() throws InterruptedException, ExecutionException {
+ TwillRunner runner = YarnTestUtils.getTwillRunner();
+ TwillController controller = runner.prepare(new SleepRunnable(600))
+ .addLogHandler(new PrinterLogHandler(new PrintWriter(System.out, true)))
+ .start();
+
+ final CountDownLatch runLatch = new CountDownLatch(1);
+ controller.addListener(new ServiceListenerAdapter() {
+ @Override
+ public void running() {
+ runLatch.countDown();
+ }
+ }, Threads.SAME_THREAD_EXECUTOR);
+
+ // Wait for application running
+ Assert.assertTrue(runLatch.await(60, TimeUnit.SECONDS));
+
+ // Find the app master ZK session and expire it two times, 10 seconds apart.
+ for (int i = 0; i < 2; i++) {
+ Assert.assertTrue(expireAppMasterZKSession(controller, 10, TimeUnit.SECONDS));
+
+ ListenableFuture<Service.State> completion = Services.getCompletionFuture(controller);
+ try {
+ completion.get(10, TimeUnit.SECONDS);
+ Assert.fail("Unexpected application termination.");
+ } catch (TimeoutException e) {
+ // OK, expected.
+ }
+ }
+
+ controller.stopAndWait();
+ }
+
+ private boolean expireAppMasterZKSession(TwillController controller, long timeout, TimeUnit timeoutUnit) {
+ MBeanServer mbeanServer = MBeanRegistry.getInstance().getPlatformMBeanServer();
+ QueryExp query = Query.isInstanceOf(new StringValueExp(ConnectionMXBean.class.getName()));
+
+ Stopwatch stopwatch = new Stopwatch();
+
+ do {
+ // Find the AM session and expire it
+ Set<ObjectName> connectionBeans = mbeanServer.queryNames(ObjectName.WILDCARD, query);
+ for (ObjectName objectName : connectionBeans) {
+
+ ConnectionMXBean connectionBean = MBeanServerInvocationHandler.newProxyInstance(mbeanServer, objectName,
+ ConnectionMXBean.class, false);
+ for (String node : connectionBean.getEphemeralNodes()) {
+ if (node.endsWith("/instances/" + controller.getRunId().getId())) {
+ // This is the AM, expire the session.
+ LOG.info("Kill AM session {}", connectionBean.getSessionId());
+ connectionBean.terminateSession();
+ return true;
+ }
+ }
+ }
+ } while (stopwatch.elapsedTime(timeoutUnit) < timeout);
+
+ return false;
+ }
+
+ /**
+ * A TwillRunnable that just keep sleeping.
+ */
+ public static final class SleepRunnable extends AbstractTwillRunnable {
+
+ private Thread runThread;
+
+ public SleepRunnable(long sleepSeconds) {
+ super(ImmutableMap.of("sleepSeconds", Long.toString(sleepSeconds)));
+ }
+
+ @Override
+ public void stop() {
+ if (runThread != null) {
+ runThread.interrupt();
+ }
+ }
+
+ @Override
+ public void run() {
+ try {
+ runThread = Thread.currentThread();
+ TimeUnit.SECONDS.sleep(Long.parseLong(getArgument("sleepSeconds")));
+ } catch (InterruptedException e) {
+ // Ignore
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/3d3c0e92/twill-yarn/src/test/java/org/apache/twill/yarn/YarnTestSuite.java
----------------------------------------------------------------------
diff --git a/twill-yarn/src/test/java/org/apache/twill/yarn/YarnTestSuite.java b/twill-yarn/src/test/java/org/apache/twill/yarn/YarnTestSuite.java
index bed613a..56fb6a5 100644
--- a/twill-yarn/src/test/java/org/apache/twill/yarn/YarnTestSuite.java
+++ b/twill-yarn/src/test/java/org/apache/twill/yarn/YarnTestSuite.java
@@ -32,7 +32,8 @@ import org.junit.runners.Suite;
LocalFileTestRun.class,
FailureRestartTestRun.class,
ProvisionTimeoutTestRun.class,
- LogHandlerTestRun.class
+ LogHandlerTestRun.class,
+ SessionExpireTestRun.class
})
public final class YarnTestSuite {
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/3d3c0e92/twill-yarn/src/test/java/org/apache/twill/yarn/YarnTestUtils.java
----------------------------------------------------------------------
diff --git a/twill-yarn/src/test/java/org/apache/twill/yarn/YarnTestUtils.java b/twill-yarn/src/test/java/org/apache/twill/yarn/YarnTestUtils.java
index 079db56..909c366 100644
--- a/twill-yarn/src/test/java/org/apache/twill/yarn/YarnTestUtils.java
+++ b/twill-yarn/src/test/java/org/apache/twill/yarn/YarnTestUtils.java
@@ -110,6 +110,10 @@ public final class YarnTestUtils {
return runnerService;
}
+ public static final InMemoryZKServer getZkServer() {
+ return zkServer;
+ }
+
/**
* Creates an unstarted instance of {@link org.apache.twill.api.TwillRunnerService}.
*/