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 2017/02/28 19:45:15 UTC
[1/9] twill git commit: Refactoring 1
Repository: twill
Updated Branches:
refs/heads/feature/twill-186 [created] eab093753
Refactoring 1
Project: http://git-wip-us.apache.org/repos/asf/twill/repo
Commit: http://git-wip-us.apache.org/repos/asf/twill/commit/572c7a79
Tree: http://git-wip-us.apache.org/repos/asf/twill/tree/572c7a79
Diff: http://git-wip-us.apache.org/repos/asf/twill/diff/572c7a79
Branch: refs/heads/feature/twill-186
Commit: 572c7a79f1c1cd3dbe8e0f50ba07a0f2ef7bdb02
Parents: 1008f2e
Author: Terence Yim <ch...@apache.org>
Authored: Mon Feb 27 21:22:32 2017 -0800
Committer: Terence Yim <ch...@apache.org>
Committed: Mon Feb 27 21:22:32 2017 -0800
----------------------------------------------------------------------
.../internal/appmaster/ApplicationMasterService.java | 10 +---------
.../twill/internal/appmaster/ExpectedContainers.java | 13 ++++++++-----
2 files changed, 9 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/twill/blob/572c7a79/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 b26b5ec..83fd0dc 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
@@ -162,7 +162,7 @@ public final class ApplicationMasterService extends AbstractYarnTwillService imp
Long.parseLong(System.getenv(EnvKeys.YARN_APP_ID_CLUSTER_TIME)),
amClient.getContainerId().toString(), getLocalizeFiles());
- this.expectedContainers = initExpectedContainers(twillSpec);
+ this.expectedContainers = new ExpectedContainers(twillSpec);
this.runningContainers = initRunningContainers(amClient.getContainerId(), amClient.getHost());
this.eventHandler = createEventHandler(twillSpec);
}
@@ -213,14 +213,6 @@ public final class ApplicationMasterService extends AbstractYarnTwillService imp
twillSpec.getRunnables(), twillRuntimeSpec.getMaxRetries());
}
- private ExpectedContainers initExpectedContainers(TwillSpecification twillSpec) {
- Map<String, Integer> expectedCounts = Maps.newHashMap();
- for (RuntimeSpecification runtimeSpec : twillSpec.getRunnables().values()) {
- expectedCounts.put(runtimeSpec.getName(), runtimeSpec.getResourceSpecification().getInstances());
- }
- return new ExpectedContainers(expectedCounts);
- }
-
@Override
public ResourceReport get() {
return runningContainers.getResourceReport();
http://git-wip-us.apache.org/repos/asf/twill/blob/572c7a79/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/ExpectedContainers.java
----------------------------------------------------------------------
diff --git a/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/ExpectedContainers.java b/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/ExpectedContainers.java
index f4ebbd0..c0ffdb0 100644
--- a/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/ExpectedContainers.java
+++ b/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/ExpectedContainers.java
@@ -19,6 +19,8 @@ package org.apache.twill.internal.appmaster;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
+import org.apache.twill.api.RuntimeSpecification;
+import org.apache.twill.api.TwillSpecification;
import java.util.Map;
@@ -30,13 +32,14 @@ final class ExpectedContainers {
private final Map<String, ExpectedCount> expectedCounts;
- ExpectedContainers(Map<String, Integer> expected) {
- expectedCounts = Maps.newHashMap();
+ ExpectedContainers(TwillSpecification twillSpec) {
+ Map<String, ExpectedCount> expectedCounts = Maps.newHashMap();
long now = System.currentTimeMillis();
-
- for (Map.Entry<String, Integer> entry : expected.entrySet()) {
- expectedCounts.put(entry.getKey(), new ExpectedCount(entry.getValue(), now));
+ for (RuntimeSpecification runtimeSpec : twillSpec.getRunnables().values()) {
+ expectedCounts.put(runtimeSpec.getName(),
+ new ExpectedCount(runtimeSpec.getResourceSpecification().getInstances(), now));
}
+ this.expectedCounts = expectedCounts;
}
synchronized void setExpected(String runnable, int expected) {
[2/9] twill git commit: Remove unnecessary throwables.propagate
Posted by ch...@apache.org.
Remove unnecessary throwables.propagate
Project: http://git-wip-us.apache.org/repos/asf/twill/repo
Commit: http://git-wip-us.apache.org/repos/asf/twill/commit/cc50d47d
Tree: http://git-wip-us.apache.org/repos/asf/twill/tree/cc50d47d
Diff: http://git-wip-us.apache.org/repos/asf/twill/diff/cc50d47d
Branch: refs/heads/feature/twill-186
Commit: cc50d47de8421649566ebcb0fb474204c6346081
Parents: 572c7a7
Author: Terence Yim <ch...@apache.org>
Authored: Mon Feb 27 21:28:03 2017 -0800
Committer: Terence Yim <ch...@apache.org>
Committed: Mon Feb 27 21:28:03 2017 -0800
----------------------------------------------------------------------
.../appmaster/ApplicationMasterService.java | 37 +++++++-------------
1 file changed, 13 insertions(+), 24 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/twill/blob/cc50d47d/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 83fd0dc..dd08cdc 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
@@ -21,7 +21,6 @@ import com.google.common.base.Preconditions;
import com.google.common.base.Predicate;
import com.google.common.base.Strings;
import com.google.common.base.Supplier;
-import com.google.common.base.Throwables;
import com.google.common.collect.DiscreteDomains;
import com.google.common.collect.HashMultiset;
import com.google.common.collect.ImmutableSet;
@@ -38,7 +37,6 @@ import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
-
import org.apache.hadoop.io.Text;
import org.apache.hadoop.security.Credentials;
import org.apache.hadoop.security.UserGroupInformation;
@@ -104,7 +102,6 @@ import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
-
import javax.annotation.Nullable;
/**
@@ -182,22 +179,18 @@ public final class ApplicationMasterService extends AbstractYarnTwillService imp
@SuppressWarnings("unchecked")
@Nullable
- private EventHandler createEventHandler(TwillSpecification twillSpec) {
- try {
- // Should be able to load by this class ClassLoader, as they packaged in the same jar.
- EventHandlerSpecification handlerSpec = twillSpec.getEventHandler();
- if (handlerSpec == null) {
- return null;
- }
-
- Class<?> handlerClass = getClass().getClassLoader().loadClass(handlerSpec.getClassName());
- Preconditions.checkArgument(EventHandler.class.isAssignableFrom(handlerClass),
- "Class {} does not implements {}",
- handlerClass, EventHandler.class.getName());
- return Instances.newInstance((Class<? extends EventHandler>) handlerClass);
- } catch (Exception e) {
- throw Throwables.propagate(e);
+ private EventHandler createEventHandler(TwillSpecification twillSpec) throws ClassNotFoundException {
+ // Should be able to load by this class ClassLoader, as they packaged in the same jar.
+ EventHandlerSpecification handlerSpec = twillSpec.getEventHandler();
+ if (handlerSpec == null) {
+ return null;
}
+
+ Class<?> handlerClass = getClass().getClassLoader().loadClass(handlerSpec.getClassName());
+ Preconditions.checkArgument(EventHandler.class.isAssignableFrom(handlerClass),
+ "Class {} does not implements {}",
+ handlerClass, EventHandler.class.getName());
+ return Instances.newInstance((Class<? extends EventHandler>) handlerClass);
}
private RunningContainers initRunningContainers(ContainerId appMasterContainerId,
@@ -707,17 +700,15 @@ public final class ApplicationMasterService extends AbstractYarnTwillService imp
}
}
- private List<LocalFile> getLocalizeFiles() {
+ private List<LocalFile> getLocalizeFiles() throws IOException {
try (Reader reader = Files.newBufferedReader(Paths.get(Constants.Files.LOCALIZE_FILES), StandardCharsets.UTF_8)) {
return new GsonBuilder().registerTypeAdapter(LocalFile.class, new LocalFileCodec())
.create().fromJson(reader, new TypeToken<List<LocalFile>>() {
}.getType());
- } catch (IOException e) {
- throw Throwables.propagate(e);
}
}
- private Map<String, Map<String, String>> getEnvironments() {
+ private Map<String, Map<String, String>> getEnvironments() throws IOException {
Path envFile = Paths.get(Constants.Files.RUNTIME_CONFIG_JAR, Constants.Files.ENVIRONMENTS);
if (!Files.exists(envFile)) {
return new HashMap<>();
@@ -726,8 +717,6 @@ public final class ApplicationMasterService extends AbstractYarnTwillService imp
try (Reader reader = Files.newBufferedReader(envFile, StandardCharsets.UTF_8)) {
return new Gson().fromJson(reader, new TypeToken<Map<String, Map<String, String>>>() {
}.getType());
- } catch (IOException e) {
- throw Throwables.propagate(e);
}
}
[8/9] twill git commit: Get rid of the inner loop in the doRun method
Posted by ch...@apache.org.
Get rid of the inner loop in the doRun method
- The inner loop can block the heartbeat thread for too long if there are a lot of runnable instances to stop
Project: http://git-wip-us.apache.org/repos/asf/twill/repo
Commit: http://git-wip-us.apache.org/repos/asf/twill/commit/aa64afae
Tree: http://git-wip-us.apache.org/repos/asf/twill/tree/aa64afae
Diff: http://git-wip-us.apache.org/repos/asf/twill/diff/aa64afae
Branch: refs/heads/feature/twill-186
Commit: aa64afae5428c9e1ea5490e3e86f337cd9efefc0
Parents: 720ed90
Author: Terence Yim <ch...@apache.org>
Authored: Tue Feb 28 11:34:35 2017 -0800
Committer: Terence Yim <ch...@apache.org>
Committed: Tue Feb 28 11:34:35 2017 -0800
----------------------------------------------------------------------
.../appmaster/ApplicationMasterService.java | 37 +++++++++-----------
.../appmaster/RunnableContainerRequest.java | 6 ++--
2 files changed, 19 insertions(+), 24 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/twill/blob/aa64afae/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 3fed157..567b57d 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
@@ -379,6 +379,8 @@ public final class ApplicationMasterService extends AbstractYarnTwillService imp
boolean isRequestRelaxed = false;
long nextTimeoutCheck = System.currentTimeMillis() + Constants.PROVISION_TIMEOUT;
while (!stopped) {
+ TimeUnit.SECONDS.sleep(1);
+
// Call allocate. It has to be made at first in order to be able to get cluster resource availability.
amClient.allocate(0.0f, allocateHandler);
@@ -389,28 +391,22 @@ 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
- if (--count <= 0) {
- break;
- }
+ if (provisioning.isEmpty() && currentRequest == null && !runnableContainerRequests.isEmpty()) {
+ RunnableContainerRequest containerRequest = runnableContainerRequests.peek();
+ // If the request at the head of the request queue is not yet ready, move it to the end of the queue
+ // so that it won't block requests that are already ready
+ if (!containerRequest.isReadyToBeProvisioned()) {
+ LOG.debug("Request not ready: {}", containerRequest);
+ runnableContainerRequests.add(runnableContainerRequests.poll());
continue;
}
- currentRequest = runnableContainerRequest.takeRequest();
+
+ currentRequest = containerRequest.takeRequest();
if (currentRequest == null) {
// All different types of resource request from current order is done, move to next one
// TODO: Need to handle order type as well
runnableContainerRequests.poll();
+ continue;
}
}
@@ -435,10 +431,6 @@ public final class ApplicationMasterService extends AbstractYarnTwillService imp
}
nextTimeoutCheck = checkProvisionTimeout(nextTimeoutCheck);
-
- if (isRunning()) {
- TimeUnit.SECONDS.sleep(1);
- }
}
}
@@ -940,8 +932,11 @@ public final class ApplicationMasterService extends AbstractYarnTwillService imp
}
}
+ LOG.info("All instances in {} for runnable {} are stopped. Ready to provision",
+ instancesToRemove, runnableName);
+
// set the container request to be ready
- containerRequest.setReadyToBeProvisioned(true);
+ containerRequest.setReadyToBeProvisioned();
// For all runnables that needs to re-request for containers, update the expected count timestamp
// so that the EventHandler would be triggered with the right expiration timestamp.
http://git-wip-us.apache.org/repos/asf/twill/blob/aa64afae/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/RunnableContainerRequest.java
----------------------------------------------------------------------
diff --git a/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/RunnableContainerRequest.java b/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/RunnableContainerRequest.java
index e001121..1dcffed 100644
--- a/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/RunnableContainerRequest.java
+++ b/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/RunnableContainerRequest.java
@@ -53,12 +53,12 @@ final class RunnableContainerRequest {
return orderType;
}
- public boolean isReadyToBeProvisioned() {
+ boolean isReadyToBeProvisioned() {
return isReadyToBeProvisioned;
}
- public void setReadyToBeProvisioned(boolean isProvisioned) {
- this.isReadyToBeProvisioned = isProvisioned;
+ void setReadyToBeProvisioned() {
+ this.isReadyToBeProvisioned = true;
}
/**
[5/9] twill git commit: Request multiple instances in the same request
Posted by ch...@apache.org.
Request multiple instances in the same request
Project: http://git-wip-us.apache.org/repos/asf/twill/repo
Commit: http://git-wip-us.apache.org/repos/asf/twill/commit/8eabbc1e
Tree: http://git-wip-us.apache.org/repos/asf/twill/tree/8eabbc1e
Diff: http://git-wip-us.apache.org/repos/asf/twill/diff/8eabbc1e
Branch: refs/heads/feature/twill-186
Commit: 8eabbc1e71ea4dd8779fcd795375a0181d366501
Parents: e2a03cf
Author: Terence Yim <ch...@apache.org>
Authored: Tue Feb 28 00:38:52 2017 -0800
Committer: Terence Yim <ch...@apache.org>
Committed: Tue Feb 28 00:38:52 2017 -0800
----------------------------------------------------------------------
.../twill/internal/appmaster/ApplicationMasterService.java | 8 +-------
1 file changed, 1 insertion(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/twill/blob/8eabbc1e/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 e2fb0e9..be43856 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
@@ -488,9 +488,7 @@ public final class ApplicationMasterService extends AbstractYarnTwillService imp
for (Multiset.Entry<String> entry : restartRunnables.entrySet()) {
LOG.info("Re-request container for {} with {} instances.", entry.getElement(), entry.getCount());
- for (int i = 0; i < entry.getCount(); i++) {
- runnableContainerRequests.add(createRunnableContainerRequest(entry.getElement()));
- }
+ runnableContainerRequests.add(createRunnableContainerRequest(entry.getElement(), entry.getCount()));
}
// For all runnables that needs to re-request for containers, update the expected count timestamp
@@ -806,10 +804,6 @@ public final class ApplicationMasterService extends AbstractYarnTwillService imp
};
}
- private RunnableContainerRequest createRunnableContainerRequest(final String runnableName) {
- return createRunnableContainerRequest(runnableName, 1);
- }
-
private RunnableContainerRequest createRunnableContainerRequest(final String runnableName,
final int numberOfInstances) {
return createRunnableContainerRequest(runnableName, numberOfInstances, true);
[7/9] twill git commit: Further simplification of placement policy
management
Posted by ch...@apache.org.
Further simplification of placement policy management
Project: http://git-wip-us.apache.org/repos/asf/twill/repo
Commit: http://git-wip-us.apache.org/repos/asf/twill/commit/720ed906
Tree: http://git-wip-us.apache.org/repos/asf/twill/tree/720ed906
Diff: http://git-wip-us.apache.org/repos/asf/twill/diff/720ed906
Branch: refs/heads/feature/twill-186
Commit: 720ed90670f9aa9159658ff92d4960f623807f09
Parents: a4c5eb5
Author: Terence Yim <ch...@apache.org>
Authored: Tue Feb 28 10:57:16 2017 -0800
Committer: Terence Yim <ch...@apache.org>
Committed: Tue Feb 28 11:24:09 2017 -0800
----------------------------------------------------------------------
.../appmaster/ApplicationMasterService.java | 14 ++++++-------
.../appmaster/PlacementPolicyManager.java | 21 ++++----------------
2 files changed, 11 insertions(+), 24 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/twill/blob/720ed906/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 893769c..3fed157 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
@@ -456,12 +456,12 @@ public final class ApplicationMasterService extends AbstractYarnTwillService imp
//Check the placement policy
String runnableName = allocationSpec.getRunnableName();
- if (!placementPolicyManager.isDistributed(runnableName)) {
+ if (!placementPolicyManager.getDistributedRunnables().contains(runnableName)) {
return;
}
//Update blacklist with hosts which are running DISTRIBUTED runnables
- for (String runnable : placementPolicyManager.getFellowRunnables(request.getKey().getRunnableName())) {
+ for (String runnable : placementPolicyManager.getFellowRunnables(runnableName)) {
for (ContainerInfo containerInfo : runningContainers.getContainerInfo(runnable)) {
// 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
@@ -567,10 +567,10 @@ public final class ApplicationMasterService extends AbstractYarnTwillService imp
Queue<RunnableContainerRequest> requests = new ConcurrentLinkedQueue<>();
// For each order in the twillSpec, create container request for runnables, depending on Placement policy.
for (TwillSpecification.Order order : twillSpec.getOrders()) {
- Set<String> distributedRunnables = placementPolicyManager.getDistributedRunnables(order.getNames());
- Set<String> defaultRunnables = Sets.newHashSet();
- defaultRunnables.addAll(order.getNames());
- defaultRunnables.removeAll(distributedRunnables);
+ Set<String> distributedRunnables = Sets.intersection(placementPolicyManager.getDistributedRunnables(),
+ order.getNames());
+ Set<String> defaultRunnables = Sets.difference(order.getNames(), distributedRunnables);
+
Map<AllocationSpecification, Collection<RuntimeSpecification>> requestsMap = Maps.newHashMap();
for (String runnableName : distributedRunnables) {
RuntimeSpecification runtimeSpec = twillSpec.getRunnables().get(runnableName);
@@ -822,7 +822,7 @@ public final class ApplicationMasterService extends AbstractYarnTwillService imp
Resource capability = createCapability(runtimeSpec.getResourceSpecification());
Map<AllocationSpecification, Collection<RuntimeSpecification>> requestsMap = Maps.newHashMap();
- if (placementPolicyManager.isDistributed(runnableName)) {
+ if (placementPolicyManager.getDistributedRunnables().contains(runnableName)) {
for (int instanceId = 0; instanceId < numberOfInstances; instanceId++) {
AllocationSpecification allocationSpecification =
new AllocationSpecification(capability, AllocationSpecification.Type.ALLOCATE_ONE_INSTANCE_AT_A_TIME,
http://git-wip-us.apache.org/repos/asf/twill/blob/720ed906/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/PlacementPolicyManager.java
----------------------------------------------------------------------
diff --git a/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/PlacementPolicyManager.java b/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/PlacementPolicyManager.java
index 95575d2..0b0e53a 100644
--- a/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/PlacementPolicyManager.java
+++ b/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/PlacementPolicyManager.java
@@ -18,8 +18,6 @@
package org.apache.twill.internal.appmaster;
-import com.google.common.base.Predicates;
-import com.google.common.collect.Sets;
import org.apache.twill.api.TwillSpecification;
import java.util.Collections;
@@ -51,22 +49,11 @@ final class PlacementPolicyManager {
}
/**
- * Given a set of runnables, get all runnables which belong to DISTRIBUTED placement policies.
- * @param givenRunnables Set of runnables.
- * @return Subset of runnables, which belong to DISTRIBUTED placement policies.
+ * Returns all runnables which belong to DISTRIBUTED placement policies.
*/
- Set<String> getDistributedRunnables(Set<String> givenRunnables) {
- return Sets.filter(policyTypeToRunnables.get(TwillSpecification.PlacementPolicy.Type.DISTRIBUTED),
- Predicates.in(givenRunnables));
- }
-
- /**
- * Checks if the placement policy is {@link TwillSpecification.PlacementPolicy.Type#DISTRIBUTED} for a given runnable.
- * @param runnableName Name of runnable.
- * @return {@code true} if the policy type is {@code DISTRIBUTED}
- */
- boolean isDistributed(String runnableName) {
- return policyTypeToRunnables.get(TwillSpecification.PlacementPolicy.Type.DISTRIBUTED).contains(runnableName);
+ Set<String> getDistributedRunnables() {
+ Set<String> runnables = policyTypeToRunnables.get(TwillSpecification.PlacementPolicy.Type.DISTRIBUTED);
+ return runnables == null ? Collections.<String>emptySet() : runnables;
}
/**
[3/9] twill git commit: Remove unnecessary intermediate method
Posted by ch...@apache.org.
Remove unnecessary intermediate method
Project: http://git-wip-us.apache.org/repos/asf/twill/repo
Commit: http://git-wip-us.apache.org/repos/asf/twill/commit/209f3ce2
Tree: http://git-wip-us.apache.org/repos/asf/twill/tree/209f3ce2
Diff: http://git-wip-us.apache.org/repos/asf/twill/diff/209f3ce2
Branch: refs/heads/feature/twill-186
Commit: 209f3ce2eb4596bdd0f675b23a78f189e8e067a2
Parents: cc50d47
Author: Terence Yim <ch...@apache.org>
Authored: Mon Feb 27 21:29:32 2017 -0800
Committer: Terence Yim <ch...@apache.org>
Committed: Mon Feb 27 21:29:32 2017 -0800
----------------------------------------------------------------------
.../internal/appmaster/ApplicationMasterService.java | 15 +++------------
1 file changed, 3 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/twill/blob/209f3ce2/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 dd08cdc..3044098 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
@@ -921,18 +921,9 @@ public final class ApplicationMasterService extends AbstractYarnTwillService imp
/**
* Helper method to restart instances of runnables.
*/
- private void restartRunnableInstances(String runnableName, @Nullable Set<Integer> instanceIds,
+ private void restartRunnableInstances(final String runnableName, @Nullable final Set<Integer> instanceIds,
final Runnable completion) {
- Runnable restartInstancesRunnable = createRestartInstancesRunnable(runnableName, instanceIds, completion);
- instanceChangeExecutor.execute(restartInstancesRunnable);
- }
-
- /**
- * Creates a Runnable for execution of restart instances request.
- */
- private Runnable createRestartInstancesRunnable(final String runnableName, @Nullable final Set<Integer> instanceIds,
- final Runnable completion) {
- return new Runnable() {
+ instanceChangeExecutor.execute(new Runnable() {
@Override
public void run() {
LOG.debug("Begin restart runnable {} instances.", runnableName);
@@ -966,7 +957,7 @@ public final class ApplicationMasterService extends AbstractYarnTwillService imp
completion.run();
}
- };
+ });
}
/**
[4/9] twill git commit: Better logging
Posted by ch...@apache.org.
Better logging
Project: http://git-wip-us.apache.org/repos/asf/twill/repo
Commit: http://git-wip-us.apache.org/repos/asf/twill/commit/e2a03cf1
Tree: http://git-wip-us.apache.org/repos/asf/twill/tree/e2a03cf1
Diff: http://git-wip-us.apache.org/repos/asf/twill/diff/e2a03cf1
Branch: refs/heads/feature/twill-186
Commit: e2a03cf10e8dac0c654adfe1d1322dd541f11d74
Parents: 209f3ce
Author: Terence Yim <ch...@apache.org>
Authored: Mon Feb 27 22:38:11 2017 -0800
Committer: Terence Yim <ch...@apache.org>
Committed: Mon Feb 27 22:38:11 2017 -0800
----------------------------------------------------------------------
.../twill/internal/appmaster/ApplicationMasterService.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/twill/blob/e2a03cf1/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 3044098..e2fb0e9 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
@@ -648,14 +648,14 @@ public final class ApplicationMasterService extends AbstractYarnTwillService imp
private void launchRunnable(List<? extends ProcessLauncher<YarnContainerInfo>> launchers,
Queue<ProvisionRequest> provisioning) {
for (ProcessLauncher<YarnContainerInfo> processLauncher : launchers) {
- LOG.info("Got container {}", processLauncher.getContainerInfo().getId());
+ LOG.info("Container allocated: {}", processLauncher.getContainerInfo().getContainer());
ProvisionRequest provisionRequest = provisioning.peek();
if (provisionRequest == null) {
continue;
}
String runnableName = provisionRequest.getRuntimeSpec().getName();
- LOG.info("Starting runnable {} with {}", runnableName, processLauncher);
+ LOG.info("Starting runnable {} in {}", runnableName, processLauncher.getContainerInfo().getContainer());
LOG.debug("Log level for Twill runnable {} is {}", runnableName,
twillRuntimeSpec.getLogLevels().get(runnableName).get(Logger.ROOT_LOGGER_NAME));
[9/9] twill git commit: Expose container instanceId instead of
parsing it from runId
Posted by ch...@apache.org.
Expose container instanceId instead of parsing it from runId
Project: http://git-wip-us.apache.org/repos/asf/twill/repo
Commit: http://git-wip-us.apache.org/repos/asf/twill/commit/eab09375
Tree: http://git-wip-us.apache.org/repos/asf/twill/tree/eab09375
Diff: http://git-wip-us.apache.org/repos/asf/twill/diff/eab09375
Branch: refs/heads/feature/twill-186
Commit: eab09375320f52d67cb5e7f092d61955df2362ae
Parents: aa64afa
Author: Terence Yim <ch...@apache.org>
Authored: Tue Feb 28 11:45:06 2017 -0800
Committer: Terence Yim <ch...@apache.org>
Committed: Tue Feb 28 11:45:06 2017 -0800
----------------------------------------------------------------------
.../twill/internal/TwillContainerController.java | 10 +++++++++-
.../apache/twill/internal/TwillContainerLauncher.java | 5 +++++
.../twill/internal/appmaster/RunningContainers.java | 13 +++----------
3 files changed, 17 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/twill/blob/eab09375/twill-core/src/main/java/org/apache/twill/internal/TwillContainerController.java
----------------------------------------------------------------------
diff --git a/twill-core/src/main/java/org/apache/twill/internal/TwillContainerController.java b/twill-core/src/main/java/org/apache/twill/internal/TwillContainerController.java
index 692e6b2..15689f5 100644
--- a/twill-core/src/main/java/org/apache/twill/internal/TwillContainerController.java
+++ b/twill-core/src/main/java/org/apache/twill/internal/TwillContainerController.java
@@ -22,6 +22,8 @@ import com.google.common.util.concurrent.Service;
import org.apache.twill.api.ServiceController;
import org.apache.twill.internal.state.Message;
+import javax.annotation.Nullable;
+
/**
* A {@link ServiceController} that allows sending a message directly. Internal use only.
*/
@@ -36,7 +38,13 @@ public interface TwillContainerController extends ServiceController, Service {
void completed(int exitStatus);
/**
- * @returns the container's live node data.
+ * @return the container's live node data.
*/
+ @Nullable
ContainerLiveNodeData getLiveNodeData();
+
+ /**
+ * @return the instance ID of the runnable that running in the container controlled by this controller.
+ */
+ int getInstanceId();
}
http://git-wip-us.apache.org/repos/asf/twill/blob/eab09375/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 8dce91e..9b6384c 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
@@ -262,6 +262,11 @@ public final class TwillContainerLauncher {
processController.cancel();
}
+ @Override
+ public int getInstanceId() {
+ return instanceId;
+ }
+
private void killAndWait(int maxWaitSecs) {
Stopwatch watch = new Stopwatch();
watch.start();
http://git-wip-us.apache.org/repos/asf/twill/blob/eab09375/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 563913b..a950c46 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,7 +37,6 @@ 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;
@@ -81,7 +80,6 @@ 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;
/**
@@ -238,7 +236,7 @@ final class RunningContainers {
try {
// Find the controller with particular instance id.
for (Map.Entry<String, TwillContainerController> entry : containers.row(runnableName).entrySet()) {
- if (getInstanceId(entry.getValue().getRunId()) == instanceId) {
+ if (entry.getValue().getInstanceId() == instanceId) {
containerId = entry.getKey();
controller = entry.getValue();
break;
@@ -451,7 +449,7 @@ final class RunningContainers {
for (Map.Entry<String, TwillContainerController> completedEntry : lookup.entrySet()) {
TwillContainerController controller = completedEntry.getValue();
- instanceId = getInstanceId(controller.getRunId());
+ instanceId = controller.getInstanceId();
// 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?
@@ -468,7 +466,7 @@ final class RunningContainers {
}
// TODO: should we remove the completed instance from instanceId and resource report even on failures?
// TODO: won't they get added back when the container is re-requested?
- removeInstanceId(runnableName, getInstanceId(controller.getRunId()));
+ removeInstanceId(runnableName, controller.getInstanceId());
resourceReport.removeRunnableResources(runnableName, containerId);
}
@@ -653,11 +651,6 @@ final class RunningContainers {
return RunIds.fromString(baseId.getId() + '-' + instanceId);
}
- private int getInstanceId(RunId runId) {
- String id = runId.getId();
- return Integer.parseInt(id.substring(id.lastIndexOf('-') + 1));
- }
-
/**
* Given the containerId, removes the corresponding containerInfo.
* @param containerId Id for the container to be removed.
[6/9] twill git commit: Refactory/simiply placement policy related
code
Posted by ch...@apache.org.
Refactory/simiply placement policy related code
Project: http://git-wip-us.apache.org/repos/asf/twill/repo
Commit: http://git-wip-us.apache.org/repos/asf/twill/commit/a4c5eb55
Tree: http://git-wip-us.apache.org/repos/asf/twill/tree/a4c5eb55
Diff: http://git-wip-us.apache.org/repos/asf/twill/diff/a4c5eb55
Branch: refs/heads/feature/twill-186
Commit: a4c5eb552887d2057bf07cbc16ccbbe5b7f47d2a
Parents: 8eabbc1
Author: Terence Yim <ch...@apache.org>
Authored: Tue Feb 28 10:45:07 2017 -0800
Committer: Terence Yim <ch...@apache.org>
Committed: Tue Feb 28 10:45:07 2017 -0800
----------------------------------------------------------------------
.../appmaster/ApplicationMasterService.java | 66 ++++++++---------
.../appmaster/PlacementPolicyManager.java | 77 +++++++++-----------
2 files changed, 66 insertions(+), 77 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/twill/blob/a4c5eb55/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 be43856..893769c 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
@@ -449,28 +449,26 @@ public final class ApplicationMasterService extends AbstractYarnTwillService imp
amClient.clearBlacklist();
//Check the allocation strategy
- AllocationSpecification currentAllocationSpecification = request.getKey();
- if (currentAllocationSpecification.getType().equals(AllocationSpecification.Type.ALLOCATE_ONE_INSTANCE_AT_A_TIME)) {
-
- //Check the placement policy
- TwillSpecification.PlacementPolicy placementPolicy =
- placementPolicyManager.getPlacementPolicy(currentAllocationSpecification.getRunnableName());
- if (placementPolicy != null
- && placementPolicy.getType().equals(TwillSpecification.PlacementPolicy.Type.DISTRIBUTED)) {
-
- //Update blacklist with hosts which are running DISTRIBUTED runnables
- for (String runnable : placementPolicyManager.getFellowRunnables(request.getKey().getRunnableName())) {
- Collection<ContainerInfo> containerStats =
- runningContainers.getContainerInfo(runnable);
- for (ContainerInfo containerInfo : containerStats) {
- // 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());
- }
- }
+ AllocationSpecification allocationSpec = request.getKey();
+ if (!allocationSpec.getType().equals(AllocationSpecification.Type.ALLOCATE_ONE_INSTANCE_AT_A_TIME)) {
+ return;
+ }
+
+ //Check the placement policy
+ String runnableName = allocationSpec.getRunnableName();
+ if (!placementPolicyManager.isDistributed(runnableName)) {
+ return;
+ }
+
+ //Update blacklist with hosts which are running DISTRIBUTED runnables
+ for (String runnable : placementPolicyManager.getFellowRunnables(request.getKey().getRunnableName())) {
+ for (ContainerInfo containerInfo : runningContainers.getContainerInfo(runnable)) {
+ // 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());
}
}
}
@@ -622,19 +620,19 @@ public final class ApplicationMasterService extends AbstractYarnTwillService imp
//Spawning 1 instance at a time
newContainers = 1;
}
+
+ // TODO: Allow user to set priority?
+ LOG.info("Request {} containers with capability {} for runnable {}", newContainers, capability, name);
+ YarnAMClient.ContainerRequestBuilder builder = amClient.addContainerRequest(capability, newContainers);
+ builder.setPriority(0);
+
TwillSpecification.PlacementPolicy placementPolicy = placementPolicyManager.getPlacementPolicy(name);
- Set<String> hosts = Sets.newHashSet();
- Set<String> racks = Sets.newHashSet();
if (placementPolicy != null) {
- hosts = placementPolicy.getHosts();
- racks = placementPolicy.getRacks();
+ builder.addHosts(placementPolicy.getHosts())
+ .addRacks(placementPolicy.getRacks());
}
- // TODO: Allow user to set priority?
- LOG.info("Request {} containers with capability {} for runnable {}", newContainers, capability, name);
- String requestId = amClient.addContainerRequest(capability, newContainers)
- .addHosts(hosts)
- .addRacks(racks)
- .setPriority(0).apply();
+
+ String requestId = builder.apply();
provisioning.add(new ProvisionRequest(runtimeSpec, requestId, newContainers, allocationType));
}
}
@@ -823,8 +821,8 @@ public final class ApplicationMasterService extends AbstractYarnTwillService imp
RuntimeSpecification runtimeSpec = twillSpec.getRunnables().get(runnableName);
Resource capability = createCapability(runtimeSpec.getResourceSpecification());
Map<AllocationSpecification, Collection<RuntimeSpecification>> requestsMap = Maps.newHashMap();
- if (placementPolicyManager.getPlacementPolicyType(runnableName).equals(
- TwillSpecification.PlacementPolicy.Type.DISTRIBUTED)) {
+
+ if (placementPolicyManager.isDistributed(runnableName)) {
for (int instanceId = 0; instanceId < numberOfInstances; instanceId++) {
AllocationSpecification allocationSpecification =
new AllocationSpecification(capability, AllocationSpecification.Type.ALLOCATE_ONE_INSTANCE_AT_A_TIME,
http://git-wip-us.apache.org/repos/asf/twill/blob/a4c5eb55/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/PlacementPolicyManager.java
----------------------------------------------------------------------
diff --git a/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/PlacementPolicyManager.java b/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/PlacementPolicyManager.java
index 91a58bb..95575d2 100644
--- a/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/PlacementPolicyManager.java
+++ b/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/PlacementPolicyManager.java
@@ -18,21 +18,36 @@
package org.apache.twill.internal.appmaster;
+import com.google.common.base.Predicates;
import com.google.common.collect.Sets;
import org.apache.twill.api.TwillSpecification;
import java.util.Collections;
+import java.util.EnumMap;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Set;
+import javax.annotation.Nullable;
/**
* This class provides helper functions for operating on a set of Placement Policies.
*/
-public class PlacementPolicyManager {
- List<TwillSpecification.PlacementPolicy> placementPolicies;
+final class PlacementPolicyManager {
- public PlacementPolicyManager(List<TwillSpecification.PlacementPolicy> placementPolicies) {
- this.placementPolicies = placementPolicies;
+ private final Map<TwillSpecification.PlacementPolicy.Type, Set<String>> policyTypeToRunnables;
+ private final Map<String, TwillSpecification.PlacementPolicy> runnablePolicies;
+
+ PlacementPolicyManager(List<TwillSpecification.PlacementPolicy> policies) {
+ this.policyTypeToRunnables = new EnumMap<>(TwillSpecification.PlacementPolicy.Type.class);
+ this.runnablePolicies = new HashMap<>();
+
+ for (TwillSpecification.PlacementPolicy policy : policies) {
+ policyTypeToRunnables.put(policy.getType(), policy.getNames());
+ for (String runnable : policy.getNames()) {
+ runnablePolicies.put(runnable, policy);
+ }
+ }
}
/**
@@ -40,24 +55,18 @@ public class PlacementPolicyManager {
* @param givenRunnables Set of runnables.
* @return Subset of runnables, which belong to DISTRIBUTED placement policies.
*/
- public Set<String> getDistributedRunnables(Set<String> givenRunnables) {
- Set<String> distributedRunnables = getAllDistributedRunnables();
- distributedRunnables.retainAll(givenRunnables);
- return distributedRunnables;
+ Set<String> getDistributedRunnables(Set<String> givenRunnables) {
+ return Sets.filter(policyTypeToRunnables.get(TwillSpecification.PlacementPolicy.Type.DISTRIBUTED),
+ Predicates.in(givenRunnables));
}
/**
- * Given a runnable, get the type of placement policy. Returns DEFAULT if no placement policy is specified.
+ * Checks if the placement policy is {@link TwillSpecification.PlacementPolicy.Type#DISTRIBUTED} for a given runnable.
* @param runnableName Name of runnable.
- * @return Placement policy type of the runnable.
+ * @return {@code true} if the policy type is {@code DISTRIBUTED}
*/
- public TwillSpecification.PlacementPolicy.Type getPlacementPolicyType(String runnableName) {
- for (TwillSpecification.PlacementPolicy placementPolicy : placementPolicies) {
- if (placementPolicy.getNames().contains(runnableName)) {
- return placementPolicy.getType();
- }
- }
- return TwillSpecification.PlacementPolicy.Type.DEFAULT;
+ boolean isDistributed(String runnableName) {
+ return policyTypeToRunnables.get(TwillSpecification.PlacementPolicy.Type.DISTRIBUTED).contains(runnableName);
}
/**
@@ -65,13 +74,12 @@ public class PlacementPolicyManager {
* @param runnableName Name of runnable.
* @return Set of runnables, with same placement policy.
*/
- public Set<String> getFellowRunnables(String runnableName) {
- for (TwillSpecification.PlacementPolicy placementPolicy : placementPolicies) {
- if (placementPolicy.getNames().contains(runnableName)) {
- return placementPolicy.getNames();
- }
+ Set<String> getFellowRunnables(String runnableName) {
+ TwillSpecification.PlacementPolicy policy = runnablePolicies.get(runnableName);
+ if (policy == null) {
+ return Collections.emptySet();
}
- return Collections.emptySet();
+ return policyTypeToRunnables.get(policy.getType());
}
/**
@@ -80,25 +88,8 @@ public class PlacementPolicyManager {
* @param runnableName Name of runnable.
* @return Placement policy of the runnable.
*/
- public TwillSpecification.PlacementPolicy getPlacementPolicy(String runnableName) {
- for (TwillSpecification.PlacementPolicy placementPolicy : placementPolicies) {
- if (placementPolicy.getNames().contains(runnableName)) {
- return placementPolicy;
- }
- }
- return null;
- }
-
- /**
- * Gets all runnables which belong to DISTRIBUTED placement policies.
- */
- private Set<String> getAllDistributedRunnables() {
- Set<String> distributedRunnables = Sets.newHashSet();
- for (TwillSpecification.PlacementPolicy placementPolicy : placementPolicies) {
- if (placementPolicy.getType().equals(TwillSpecification.PlacementPolicy.Type.DISTRIBUTED)) {
- distributedRunnables.addAll(placementPolicy.getNames());
- }
- }
- return distributedRunnables;
+ @Nullable
+ TwillSpecification.PlacementPolicy getPlacementPolicy(String runnableName) {
+ return runnablePolicies.get(runnableName);
}
}