You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@twill.apache.org by hs...@apache.org on 2017/02/07 18:38:13 UTC
twill git commit: (TWILL-181) allow setting the maximum number of
retries per runnable
Repository: twill
Updated Branches:
refs/heads/master db7f8d643 -> 78fc26359
(TWILL-181) allow setting the maximum number of retries per runnable
Patch by @serranom
Each runnable can have a configured number of max retries. If not set, then retries are unlimited as before. Retries are scaled according to the number of instances for each Runnable.
add withMaxTries(runnableName, int) to TwillPreparer
add withMaxTries(runnableName, int) to YarnTwillPreparer. This stores a map from runnableName to maxRetries.
this map becomes part of the twillRuntimeSpecification and RuntimeSpecification interface and is added to TwillRuntimeSpecificationCodec
ApplicationMasterService.initRunningContainers is updated to pass a map of runnables to maxretries.
updated RunningContainers so that it keeps count of the number of retries per runnable and uses this in handleCompleted() to determine if it should retry. Since every instance is the same as any other, if I'm starting 10 instances of a Runnable, and wanted a max retry count of 3, then that would scale the total number of retries to 30. Each instance gets (on average) 3 tries. Since the instances are interchangeable, there is no concept of a discrete instance being retried.
updated logging to not have anything special if max wasn't set and to log the number of retries left and when they have been exhausted.
This closes #23 from Github
Signed-off-by: Henry Saputra <hs...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/twill/repo
Commit: http://git-wip-us.apache.org/repos/asf/twill/commit/78fc2635
Tree: http://git-wip-us.apache.org/repos/asf/twill/tree/78fc2635
Diff: http://git-wip-us.apache.org/repos/asf/twill/diff/78fc2635
Branch: refs/heads/master
Commit: 78fc2635992ad7fa164c97cf59ddfd8b0cb0aef9
Parents: db7f8d6
Author: martin <ma...@attivio.com>
Authored: Sat Feb 4 11:40:36 2017 -0500
Committer: Henry Saputra <hs...@apache.org>
Committed: Tue Feb 7 10:37:45 2017 -0800
----------------------------------------------------------------------
.../org/apache/twill/api/TwillPreparer.java | 10 ++
.../java/org/apache/twill/api/TwillRunner.java | 6 +-
.../internal/TwillRuntimeSpecification.java | 8 +-
.../json/TwillRuntimeSpecificationCodec.java | 12 +-
.../appmaster/ApplicationMasterService.java | 15 +-
.../internal/appmaster/RunningContainers.java | 96 +++++++++--
.../apache/twill/yarn/YarnTwillPreparer.java | 31 ++--
.../apache/twill/yarn/MaxRetriesTestRun.java | 162 +++++++++++++++++++
.../org/apache/twill/yarn/YarnTestSuite.java | 3 +-
9 files changed, 312 insertions(+), 31 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/twill/blob/78fc2635/twill-api/src/main/java/org/apache/twill/api/TwillPreparer.java
----------------------------------------------------------------------
diff --git a/twill-api/src/main/java/org/apache/twill/api/TwillPreparer.java b/twill-api/src/main/java/org/apache/twill/api/TwillPreparer.java
index 7004d89..fd568b3 100644
--- a/twill-api/src/main/java/org/apache/twill/api/TwillPreparer.java
+++ b/twill-api/src/main/java/org/apache/twill/api/TwillPreparer.java
@@ -220,6 +220,16 @@ public interface TwillPreparer {
TwillPreparer withBundlerClassAcceptor(ClassAcceptor classAcceptor);
/**
+ * Sets the maximum number of times (per instance) a runnable will be retried if it exits without success. The default
+ * behavior is to retry indefinitely.
+ *
+ * @param runnableName Name of the {@link TwillRunnable}.
+ * @param maxRetries maximum number of retries per instance
+ * @return This {@link TwillPreparer}
+ */
+ TwillPreparer withMaxRetries(String runnableName, int maxRetries);
+
+ /**
* Adds security credentials for the runtime environment to gives application access to resources.
*
* @param secureStore Contains security token available for the runtime environment.
http://git-wip-us.apache.org/repos/asf/twill/blob/78fc2635/twill-api/src/main/java/org/apache/twill/api/TwillRunner.java
----------------------------------------------------------------------
diff --git a/twill-api/src/main/java/org/apache/twill/api/TwillRunner.java b/twill-api/src/main/java/org/apache/twill/api/TwillRunner.java
index 0393a85..845ced8 100644
--- a/twill-api/src/main/java/org/apache/twill/api/TwillRunner.java
+++ b/twill-api/src/main/java/org/apache/twill/api/TwillRunner.java
@@ -46,13 +46,17 @@ public interface TwillRunner {
/**
* Prepares to run the given {@link TwillRunnable} with {@link ResourceSpecification#BASIC} resource specification.
+ * The name for the runnable will be defaulted to {@code runnable.getClass().getSimpleName()}
+ *
* @param runnable The runnable to run through Twill when {@link TwillPreparer#start()} is called.
* @return A {@link TwillPreparer} for setting up runtime options.
*/
TwillPreparer prepare(TwillRunnable runnable);
/**
- * Prepares to run the given {@link TwillRunnable} with the given resource specification.
+ * Prepares to run the given {@link TwillRunnable} with the given resource specification. The name for the runnable
+ * will be defaulted to {@code runnable.getClass().getSimpleName()}
+ *
* @param runnable The runnable to run through Twill when {@link TwillPreparer#start()} is called.
* @param resourceSpecification The resource specification for running the runnable.
* @return A {@link TwillPreparer} for setting up runtime options.
http://git-wip-us.apache.org/repos/asf/twill/blob/78fc2635/twill-core/src/main/java/org/apache/twill/internal/TwillRuntimeSpecification.java
----------------------------------------------------------------------
diff --git a/twill-core/src/main/java/org/apache/twill/internal/TwillRuntimeSpecification.java b/twill-core/src/main/java/org/apache/twill/internal/TwillRuntimeSpecification.java
index 387b01d..2f74316 100644
--- a/twill-core/src/main/java/org/apache/twill/internal/TwillRuntimeSpecification.java
+++ b/twill-core/src/main/java/org/apache/twill/internal/TwillRuntimeSpecification.java
@@ -40,11 +40,12 @@ public class TwillRuntimeSpecification {
private final int reservedMemory;
private final String rmSchedulerAddr;
private final Map<String, Map<String, String>> logLevels;
+ private final Map<String, Integer> maxRetries;
public TwillRuntimeSpecification(TwillSpecification twillSpecification, String fsUser, URI twillAppDir,
String zkConnectStr, RunId twillRunId, String twillAppName,
int reservedMemory, @Nullable String rmSchedulerAddr,
- Map<String, Map<String, String>> logLevels) {
+ Map<String, Map<String, String>> logLevels, Map<String, Integer> maxRetries) {
this.twillSpecification = twillSpecification;
this.fsUser = fsUser;
this.twillAppDir = twillAppDir;
@@ -54,6 +55,7 @@ public class TwillRuntimeSpecification {
this.reservedMemory = reservedMemory;
this.rmSchedulerAddr = rmSchedulerAddr;
this.logLevels = logLevels;
+ this.maxRetries = maxRetries;
}
public TwillSpecification getTwillSpecification() {
@@ -92,4 +94,8 @@ public class TwillRuntimeSpecification {
public Map<String, Map<String, String>> getLogLevels() {
return logLevels;
}
+
+ public Map<String, Integer> getMaxRetries() {
+ return maxRetries;
+ }
}
http://git-wip-us.apache.org/repos/asf/twill/blob/78fc2635/twill-core/src/main/java/org/apache/twill/internal/json/TwillRuntimeSpecificationCodec.java
----------------------------------------------------------------------
diff --git a/twill-core/src/main/java/org/apache/twill/internal/json/TwillRuntimeSpecificationCodec.java b/twill-core/src/main/java/org/apache/twill/internal/json/TwillRuntimeSpecificationCodec.java
index 133371d..1276555 100644
--- a/twill-core/src/main/java/org/apache/twill/internal/json/TwillRuntimeSpecificationCodec.java
+++ b/twill-core/src/main/java/org/apache/twill/internal/json/TwillRuntimeSpecificationCodec.java
@@ -48,6 +48,7 @@ final class TwillRuntimeSpecificationCodec implements JsonSerializer<TwillRuntim
private static final String RM_SCHEDULER_ADDR = "rmSchedulerAddr";
private static final String TWILL_SPEC = "twillSpecification";
private static final String LOG_LEVELS = "logLevels";
+ private static final String MAX_RETRIES = "maxRetries";
@Override
public JsonElement serialize(TwillRuntimeSpecification src, Type typeOfSrc, JsonSerializationContext context) {
@@ -65,6 +66,9 @@ final class TwillRuntimeSpecificationCodec implements JsonSerializer<TwillRuntim
context.serialize(src.getTwillSpecification(), new TypeToken<TwillSpecification>() { }.getType()));
json.add(LOG_LEVELS,
context.serialize(src.getLogLevels(), new TypeToken<Map<String, Map<String, String>>>() { }.getType()));
+ json.add(MAX_RETRIES,
+ context.serialize(src.getMaxRetries(), new TypeToken<Map<String, Integer>>() { }.getType()));
+
return json;
}
@@ -77,6 +81,9 @@ final class TwillRuntimeSpecificationCodec implements JsonSerializer<TwillRuntim
jsonObj.get(TWILL_SPEC), new TypeToken<TwillSpecification>() { }.getType());
Map<String, Map<String, String>> logLevels =
context.deserialize(jsonObj.get(LOG_LEVELS), new TypeToken<Map<String, Map<String, String>>>() { }.getType());
+ Map<String, Integer> maxRetries =
+ context.deserialize(jsonObj.get(MAX_RETRIES), new TypeToken<Map<String, Integer>>() { }.getType());
+
return new TwillRuntimeSpecification(twillSpecification,
jsonObj.get(FS_USER).getAsString(),
URI.create(jsonObj.get(TWILL_APP_DIR).getAsString()),
@@ -85,7 +92,8 @@ final class TwillRuntimeSpecificationCodec implements JsonSerializer<TwillRuntim
jsonObj.get(TWILL_APP_NAME).getAsString(),
jsonObj.get(RESERVED_MEMORY).getAsInt(),
jsonObj.has(RM_SCHEDULER_ADDR) ?
- jsonObj.get(RM_SCHEDULER_ADDR).getAsString() : null,
- logLevels);
+ jsonObj.get(RM_SCHEDULER_ADDR).getAsString() : null,
+ logLevels,
+ maxRetries);
}
}
http://git-wip-us.apache.org/repos/asf/twill/blob/78fc2635/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/ApplicationMasterService.java
----------------------------------------------------------------------
diff --git a/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/ApplicationMasterService.java b/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/ApplicationMasterService.java
index 985fad1..9ea60ae 100644
--- a/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/ApplicationMasterService.java
+++ b/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/ApplicationMasterService.java
@@ -207,8 +207,8 @@ public final class ApplicationMasterService extends AbstractYarnTwillService imp
Integer.parseInt(System.getenv(EnvKeys.YARN_CONTAINER_MEMORY_MB)),
appMasterHost, null);
String appId = appMasterContainerId.getApplicationAttemptId().getApplicationId().toString();
- return new RunningContainers(appId, appMasterResources, zkClient,
- applicationLocation, twillSpec.getRunnables().keySet());
+ return new RunningContainers(appId, appMasterResources, zkClient, applicationLocation,
+ twillSpec.getRunnables(), twillRuntimeSpec.getMaxRetries());
}
private ExpectedContainers initExpectedContainers(TwillSpecification twillSpec) {
@@ -226,8 +226,8 @@ public final class ApplicationMasterService extends AbstractYarnTwillService imp
@Override
protected void doStart() throws Exception {
- LOG.info("Start application master with spec: " +
- TwillRuntimeSpecificationAdapter.create().toJson(twillRuntimeSpec));
+ LOG.info("Start application master with spec: {}",
+ TwillRuntimeSpecificationAdapter.create().toJson(twillRuntimeSpec));
// initialize the event handler, if it fails, it will fail the application.
if (eventHandler != null) {
@@ -403,12 +403,14 @@ public final class ApplicationMasterService extends AbstractYarnTwillService imp
// If nothing is in provisioning, and no pending request, move to next one
int count = runnableContainerRequests.size();
+ LOG.debug("Runnable container requests: {}", count);
while (provisioning.isEmpty() && currentRequest == null && !runnableContainerRequests.isEmpty()) {
RunnableContainerRequest runnableContainerRequest = runnableContainerRequests.peek();
if (!runnableContainerRequest.isReadyToBeProvisioned()) {
// take it out from queue and put it back at the end for second chance.
runnableContainerRequest = runnableContainerRequests.poll();
runnableContainerRequests.add(runnableContainerRequest);
+ LOG.debug("Request not ready: {}", runnableContainerRequest);
// We checked all the requests that were pending when we started this loop
// Any remaining requests are not ready to be provisioned
@@ -477,6 +479,7 @@ public final class ApplicationMasterService extends AbstractYarnTwillService imp
// Yarn Resource Manager may include port in the node name depending on the setting
// YarnConfiguration.RM_SCHEDULER_INCLUDE_PORT_IN_NODE_NAME. It is safe to add both
// the names (with and without port) to the blacklist.
+ LOG.debug("Adding {} to host blacklist", containerInfo.getHost().getHostName());
amClient.addToBlacklist(containerInfo.getHost().getHostName());
amClient.addToBlacklist(containerInfo.getHost().getHostName() + ":" + containerInfo.getPort());
}
@@ -642,7 +645,7 @@ public final class ApplicationMasterService extends AbstractYarnTwillService imp
racks = placementPolicy.getRacks();
}
// TODO: Allow user to set priority?
- LOG.info("Request {} container with capability {} for runnable {}", newContainers, capability, name);
+ LOG.info("Request {} containers with capability {} for runnable {}", newContainers, capability, name);
String requestId = amClient.addContainerRequest(capability, newContainers)
.addHosts(hosts)
.addRacks(racks)
@@ -705,7 +708,7 @@ public final class ApplicationMasterService extends AbstractYarnTwillService imp
provisioning.poll();
}
if (expectedContainers.getExpected(runnableName) == runningContainers.count(runnableName)) {
- LOG.info("Runnable " + runnableName + " fully provisioned with " + containerCount + " instances.");
+ LOG.info("Runnable {} fully provisioned with {} instances.", runnableName, containerCount);
}
}
}
http://git-wip-us.apache.org/repos/asf/twill/blob/78fc2635/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/RunningContainers.java
----------------------------------------------------------------------
diff --git a/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/RunningContainers.java b/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/RunningContainers.java
index d0fad26..563913b 100644
--- a/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/RunningContainers.java
+++ b/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/RunningContainers.java
@@ -37,9 +37,11 @@ import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.Service;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
+
import org.apache.hadoop.yarn.api.records.ContainerState;
import org.apache.twill.api.ResourceReport;
import org.apache.twill.api.RunId;
+import org.apache.twill.api.RuntimeSpecification;
import org.apache.twill.api.TwillRunResources;
import org.apache.twill.api.logging.LogEntry;
import org.apache.twill.filesystem.Location;
@@ -79,6 +81,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
+
import javax.annotation.Nullable;
/**
@@ -113,9 +116,11 @@ final class RunningContainers {
private final Location applicationLocation;
private final Set<String> runnableNames;
private final Map<String, Map<String, String>> logLevels;
+ private final Map<String, Integer> maxRetries;
+ private final Map<String, Map<Integer, AtomicInteger>> numRetries;
RunningContainers(String appId, TwillRunResources appMasterResources, ZKClient zookeeperClient,
- Location applicationLocation, Set<String> runnableNames) {
+ Location applicationLocation, Map<String, RuntimeSpecification> runnables, Map<String, Integer> maxRetriesMap) {
containers = HashBasedTable.create();
runnableInstances = Maps.newHashMap();
completedContainerCount = Maps.newHashMap();
@@ -126,8 +131,10 @@ final class RunningContainers {
zkClient = zookeeperClient;
containerStats = HashMultimap.create();
this.applicationLocation = applicationLocation;
- this.runnableNames = runnableNames;
+ this.runnableNames = runnables.keySet();
this.logLevels = new TreeMap<>();
+ this.maxRetries = Maps.newHashMap(maxRetriesMap);
+ this.numRetries = Maps.newHashMap();
}
/**
@@ -256,6 +263,10 @@ final class RunningContainers {
if (removeContainerInfo(containerId)) {
containers.remove(runnableName, containerId);
removeInstanceId(runnableName, instanceId);
+
+ // clear the entry from numRetries since we are intentionally stopping this instance
+ numRetries.get(runnableName).remove(instanceId);
+
resourceReport.removeRunnableResources(runnableName, containerId);
containerChange.signalAll();
}
@@ -394,6 +405,7 @@ final class RunningContainers {
try {
containers.clear();
runnableInstances.clear();
+ numRetries.clear();
containerStats.clear();
} finally {
containerLock.unlock();
@@ -434,9 +446,12 @@ final class RunningContainers {
}
boolean containerStopped = false;
+ final String runnableName = lookup.keySet().iterator().next();
+ int instanceId = 0;
+
for (Map.Entry<String, TwillContainerController> completedEntry : lookup.entrySet()) {
- String runnableName = completedEntry.getKey();
TwillContainerController controller = completedEntry.getValue();
+ instanceId = getInstanceId(controller.getRunId());
// TODO: Can there be multiple controllers for a single container?
// TODO: What is the best way to determine whether to restart container when there are multiple controllers?
@@ -456,13 +471,14 @@ final class RunningContainers {
removeInstanceId(runnableName, getInstanceId(controller.getRunId()));
resourceReport.removeRunnableResources(runnableName, containerId);
}
+
if (exitStatus != ContainerExitCodes.SUCCESS) {
LOG.warn("Container {} exited abnormally with state {}, exit code {}.",
containerId, state, exitStatus);
- if (!containerStopped && shouldRetry(exitStatus)) {
+ if (!containerStopped && shouldRetry(runnableName, instanceId, exitStatus)) {
LOG.info("Re-request the container {} for exit code {}.", containerId, exitStatus);
- restartRunnables.add(lookup.keySet().iterator().next());
+ restartRunnables.add(runnableName);
} else if (containerStopped) {
LOG.info("Container {} is being stopped, will not re-request", containerId);
}
@@ -477,10 +493,30 @@ final class RunningContainers {
}
}
- private boolean shouldRetry(int exitCode) {
- return exitCode != ContainerExitCodes.SUCCESS
- && exitCode != ContainerExitCodes.DISKS_FAILED
- && exitCode != ContainerExitCodes.INIT_FAILED;
+ private boolean shouldRetry(String runnableName, int instanceId, int exitCode) {
+ boolean possiblyRetry =
+ exitCode != ContainerExitCodes.SUCCESS &&
+ exitCode != ContainerExitCodes.DISKS_FAILED &&
+ exitCode != ContainerExitCodes.INIT_FAILED;
+
+ if (possiblyRetry) {
+ int max = getMaxRetries(runnableName);
+ if (max == Integer.MAX_VALUE) {
+ return true; // retry without special log msg
+ }
+
+ final int retryCount = getRetryCount(runnableName, instanceId);
+ if (retryCount == max) {
+ LOG.info("Retries exhausted for instance {} of runnable {}.", instanceId, runnableName);
+ return false;
+ } else {
+ LOG.info("Attempting {} of {} retries for instance {} of runnable {}.",
+ retryCount + 1, max, instanceId, runnableName);
+ return true;
+ }
+ } else {
+ return false;
+ }
}
private boolean isControllerStopped(TwillContainerController controller) {
@@ -525,7 +561,15 @@ final class RunningContainers {
instances = new BitSet();
runnableInstances.put(runnableName, instances);
}
- int instanceId = instances.nextClearBit(0);
+
+ // get the next free instance that has not exceeded its maximum retries.
+ int instanceId = 0;
+ int maxRetries = getMaxRetries(runnableName);
+ while (getRetryCount(runnableName, (instanceId = instances.nextClearBit(instanceId))) == maxRetries) {
+ instanceId++;
+ }
+ // count starts at -1, so first increment makes it 0
+ incrementRetryCount(runnableName, instanceId);
instances.set(instanceId);
return instanceId;
}
@@ -563,6 +607,38 @@ final class RunningContainers {
return instances == null ? 0 : instances.cardinality();
}
+ /**
+ * Returns the maximum number of retries for the runnable.
+ */
+ private int getMaxRetries(String runnableName) {
+ int max = Integer.MAX_VALUE;
+ if (maxRetries.containsKey(runnableName)) {
+ max = maxRetries.get(runnableName);
+ }
+ return max;
+ }
+
+ /**
+ * Returns the retry count for the runnable and instance, while initializing a counter if one doesn't already exist.
+ */
+ private int getRetryCount(String runnableName, int instanceId) {
+ Map<Integer, AtomicInteger> runnableNumRetries = numRetries.get(runnableName);
+ if (runnableNumRetries == null) {
+ runnableNumRetries = Maps.newHashMap();
+ numRetries.put(runnableName, runnableNumRetries);
+ }
+ AtomicInteger cnt = runnableNumRetries.get(instanceId);
+ if (cnt == null) {
+ cnt = new AtomicInteger(-1);
+ runnableNumRetries.put(instanceId, cnt);
+ }
+ return cnt.get();
+ }
+
+ private void incrementRetryCount(String runnableName, int instanceId) {
+ numRetries.get(runnableName).get(instanceId).incrementAndGet();
+ }
+
private RunId getRunId(String runnableName, int instanceId) {
RunId baseId;
http://git-wip-us.apache.org/repos/asf/twill/blob/78fc2635/twill-yarn/src/main/java/org/apache/twill/yarn/YarnTwillPreparer.java
----------------------------------------------------------------------
diff --git a/twill-yarn/src/main/java/org/apache/twill/yarn/YarnTwillPreparer.java b/twill-yarn/src/main/java/org/apache/twill/yarn/YarnTwillPreparer.java
index 5f599d6..c11e483 100644
--- a/twill-yarn/src/main/java/org/apache/twill/yarn/YarnTwillPreparer.java
+++ b/twill-yarn/src/main/java/org/apache/twill/yarn/YarnTwillPreparer.java
@@ -20,6 +20,7 @@ package org.apache.twill.yarn;
import com.google.common.base.Function;
import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
+import com.google.common.base.Predicates;
import com.google.common.base.Throwables;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.HashMultimap;
@@ -139,18 +140,19 @@ final class YarnTwillPreparer implements TwillPreparer {
private final List<URI> resources = Lists.newArrayList();
private final List<String> classPaths = Lists.newArrayList();
private final ListMultimap<String, String> runnableArgs = ArrayListMultimap.create();
- private final Map<String, Map<String, String>> environments = new HashMap<>();
+ private final Map<String, Map<String, String>> environments = Maps.newHashMap();
private final List<String> applicationClassPaths = Lists.newArrayList();
private final Credentials credentials;
private final int reservedMemory;
private final File localStagingDir;
- private final Map<String, Map<String, String>> logLevels = new HashMap<>();
+ private final Map<String, Map<String, String>> logLevels = Maps.newHashMap();
private final LocationCache locationCache;
private final Set<URL> twillClassPaths;
private String schedulerQueue;
private String extraOptions;
private JvmOptions.DebugOptions debugOptions = JvmOptions.DebugOptions.NO_DEBUG;
private ClassAcceptor classAcceptor;
+ private final Map<String, Integer> maxRetries = Maps.newHashMap();
YarnTwillPreparer(YarnConfiguration yarnConfig, TwillSpecification twillSpec, RunId runId,
YarnAppClient yarnAppClient, String zkConnectString, Location appLocation, Set<URL> twillClassPaths,
@@ -176,6 +178,12 @@ final class YarnTwillPreparer implements TwillPreparer {
setLogLevel(LogEntry.Level.INFO);
}
+ private void confirmRunnableName(String runnableName) {
+ Preconditions.checkNotNull(runnableName);
+ Preconditions.checkArgument(twillSpec.getRunnables().containsKey(runnableName),
+ "Runnable %s is not defined in the application.", runnableName);
+ }
+
@Override
public TwillPreparer addLogHandler(LogHandler handler) {
logHandlers.add(handler);
@@ -234,8 +242,7 @@ final class YarnTwillPreparer implements TwillPreparer {
@Override
public TwillPreparer withArguments(String runnableName, Iterable<String> args) {
- Preconditions.checkArgument(twillSpec.getRunnables().containsKey(runnableName),
- "Runnable %s is not defined in the application.", runnableName);
+ confirmRunnableName(runnableName);
runnableArgs.putAll(runnableName, args);
return this;
}
@@ -284,8 +291,7 @@ final class YarnTwillPreparer implements TwillPreparer {
@Override
public TwillPreparer withEnv(String runnableName, Map<String, String> env) {
- Preconditions.checkArgument(twillSpec.getRunnables().containsKey(runnableName),
- "Runnable %s is not defined in the application.", runnableName);
+ confirmRunnableName(runnableName);
setEnv(runnableName, env, true);
return this;
}
@@ -308,6 +314,13 @@ final class YarnTwillPreparer implements TwillPreparer {
}
@Override
+ public TwillPreparer withMaxRetries(String runnableName, int maxRetries) {
+ confirmRunnableName(runnableName);
+ this.maxRetries.put(runnableName, maxRetries);
+ return this;
+ }
+
+ @Override
public TwillPreparer addSecureStore(SecureStore secureStore) {
Object store = secureStore.getStore();
Preconditions.checkArgument(store instanceof Credentials, "Only Hadoop Credentials is supported.");
@@ -331,9 +344,7 @@ final class YarnTwillPreparer implements TwillPreparer {
@Override
public TwillPreparer setLogLevels(String runnableName, Map<String, LogEntry.Level> runnableLogLevels) {
- Preconditions.checkNotNull(runnableName);
- Preconditions.checkArgument(twillSpec.getRunnables().containsKey(runnableName),
- "Runnable %s is not defined in the application.", runnableName);
+ confirmRunnableName(runnableName);
Preconditions.checkNotNull(runnableLogLevels);
Preconditions.checkArgument(!(logLevels.containsKey(Logger.ROOT_LOGGER_NAME)
&& logLevels.get(Logger.ROOT_LOGGER_NAME) == null));
@@ -616,7 +627,7 @@ final class YarnTwillPreparer implements TwillPreparer {
new TwillRuntimeSpecification(newTwillSpec, appLocation.getLocationFactory().getHomeLocation().getName(),
appLocation.toURI(), zkConnectString, runId, twillSpec.getName(),
reservedMemory, yarnConfig.get(YarnConfiguration.RM_SCHEDULER_ADDRESS),
- logLevels), writer);
+ logLevels, maxRetries), writer);
}
LOG.debug("Done {}", targetFile);
}
http://git-wip-us.apache.org/repos/asf/twill/blob/78fc2635/twill-yarn/src/test/java/org/apache/twill/yarn/MaxRetriesTestRun.java
----------------------------------------------------------------------
diff --git a/twill-yarn/src/test/java/org/apache/twill/yarn/MaxRetriesTestRun.java b/twill-yarn/src/test/java/org/apache/twill/yarn/MaxRetriesTestRun.java
new file mode 100644
index 0000000..f178c89
--- /dev/null
+++ b/twill-yarn/src/test/java/org/apache/twill/yarn/MaxRetriesTestRun.java
@@ -0,0 +1,162 @@
+/*
+ * 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 org.apache.twill.api.AbstractTwillRunnable;
+import org.apache.twill.api.ResourceSpecification;
+import org.apache.twill.api.TwillController;
+import org.apache.twill.api.TwillRunner;
+import org.apache.twill.api.logging.LogEntry;
+import org.apache.twill.api.logging.LogHandler;
+import org.apache.twill.api.logging.PrinterLogHandler;
+import org.junit.Assert;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.PrintWriter;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Unit tests for RunningContainers class.
+ */
+public final class MaxRetriesTestRun extends BaseYarnTest {
+
+ @Test
+ public void maxRetriesOneInstance() throws TimeoutException, ExecutionException {
+ maxRetriesRun(1);
+ }
+
+ @Test
+ public void maxRetriesTwoInstances() throws TimeoutException, ExecutionException {
+ maxRetriesRun(2);
+ }
+
+ private void maxRetriesRun(final int instances) throws TimeoutException, ExecutionException {
+ TwillRunner runner = getTwillRunner();
+ final int maxRetries = 3;
+ final AtomicInteger retriesSeen = new AtomicInteger(0);
+
+ ResourceSpecification resource = ResourceSpecification.Builder.with().setVirtualCores(1)
+ .setMemory(512, ResourceSpecification.SizeUnit.MEGA).setInstances(instances).build();
+
+ TwillController controller = runner.prepare(new FailingServer(), resource)
+ .withMaxRetries(FailingServer.class.getSimpleName(), maxRetries)
+ .addLogHandler(new PrinterLogHandler(new PrintWriter(System.out, true)))
+ .addLogHandler(new LogHandler() {
+ @Override
+ public void onLog(LogEntry logEntry) {
+ if (logEntry.getMessage().contains("retries for instance")) {
+ retriesSeen.incrementAndGet();
+ }
+ }
+ })
+ .start();
+
+ controller.awaitTerminated(2, TimeUnit.MINUTES);
+ Assert.assertEquals(maxRetries * instances, retriesSeen.get());
+ }
+
+ @Test
+ public void maxRetriesWithIncreasedInstances() throws InterruptedException, ExecutionException {
+ TwillRunner runner = getTwillRunner();
+ final int maxRetries = 3;
+ final AtomicInteger retriesSeen = new AtomicInteger(0);
+ final CountDownLatch retriesExhausted = new CountDownLatch(1);
+ final CountDownLatch allRunning = new CountDownLatch(1);
+
+ // start with 2 instances
+ ResourceSpecification resource = ResourceSpecification.Builder.with().setVirtualCores(1)
+ .setMemory(512, ResourceSpecification.SizeUnit.MEGA).setInstances(2).build();
+
+ TwillController controller = runner.prepare(new FailingInstanceServer(), resource)
+ .withMaxRetries(FailingInstanceServer.class.getSimpleName(), maxRetries)
+ .addLogHandler(new PrinterLogHandler(new PrintWriter(System.out, true))).addLogHandler(new LogHandler() {
+ @Override
+ public void onLog(LogEntry logEntry) {
+ if (logEntry.getMessage().contains(" retries for instance")) {
+ retriesSeen.incrementAndGet();
+ }
+ if (logEntry.getMessage().contains("Retries exhausted")) {
+ retriesExhausted.countDown();
+ }
+ if (logEntry.getMessage().contains("fully provisioned with 2 instances")) {
+ allRunning.countDown();
+ }
+ }
+ }).start();
+
+ try {
+ // wait for initial instances to have started
+ allRunning.await();
+
+ /*
+ * now increase the number of instances. these should fail since there instance ids are > 1. afterwards, the
+ * number of retries should be 3 since only this one instance failed.
+ */
+ controller.changeInstances(FailingInstanceServer.class.getSimpleName(), 3);
+ retriesExhausted.await();
+ Assert.assertEquals(3, retriesSeen.get());
+
+ } finally {
+ controller.terminate().get();
+ }
+ }
+
+ /**
+ * A runnable that throws an exception during run().
+ */
+ public static final class FailingServer extends AbstractTwillRunnable {
+ private static final Logger LOG = LoggerFactory.getLogger(FailingServer.class);
+
+ @Override
+ public void run() {
+ LOG.info("Failing instance {}", getContext().getInstanceId());
+ throw new RuntimeException("FAIL early FAIL often");
+ }
+ }
+
+ /**
+ * A runnable that throws an exception during run() but only for instance ids > 1.
+ */
+ public static final class FailingInstanceServer extends AbstractTwillRunnable {
+ private static final Logger LOG = LoggerFactory.getLogger(FailingServer.class);
+
+ @Override
+ public void run() {
+ if (getContext().getInstanceId() > 1) {
+ LOG.info("Failing instance {}", getContext().getInstanceId());
+ throw new RuntimeException("FAIL early FAIL often");
+ } else {
+ LOG.info("Instance {} is running", getContext().getInstanceId());
+ while (true) {
+ try {
+ Thread.sleep(100);
+ } catch (InterruptedException e) {
+ return;
+ }
+ }
+ }
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/twill/blob/78fc2635/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 d1ba134..56172da 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
@@ -42,7 +42,8 @@ import org.junit.runners.Suite;
ServiceDiscoveryTestRun.class,
SessionExpireTestRun.class,
TaskCompletedTestRun.class,
- RestartRunnableTestRun.class
+ RestartRunnableTestRun.class,
+ MaxRetriesTestRun.class
})
public final class YarnTestSuite extends BaseYarnTest {
}