You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ambari.apache.org by jo...@apache.org on 2016/11/14 15:46:28 UTC
[05/19] ambari git commit: AMBARI-18569. Execute topology tasks in
parallel by hosts. (Attila Doroszlai via stoader)
AMBARI-18569. Execute topology tasks in parallel by hosts. (Attila Doroszlai via stoader)
Project: http://git-wip-us.apache.org/repos/asf/ambari/repo
Commit: http://git-wip-us.apache.org/repos/asf/ambari/commit/5dea886e
Tree: http://git-wip-us.apache.org/repos/asf/ambari/tree/5dea886e
Diff: http://git-wip-us.apache.org/repos/asf/ambari/diff/5dea886e
Branch: refs/heads/branch-feature-AMBARI-18456
Commit: 5dea886efde58fb93033f705b5dd43b5ccd72ad6
Parents: 2ddd7a3
Author: Attila Doroszlai <ad...@hortonworks.com>
Authored: Fri Nov 11 12:22:04 2016 +0100
Committer: Toader, Sebastian <st...@hortonworks.com>
Committed: Fri Nov 11 12:22:04 2016 +0100
----------------------------------------------------------------------
.../server/configuration/Configuration.java | 37 ++++++++++++++-
.../AmbariManagementControllerImpl.java | 20 ++++----
.../internal/ComponentResourceProvider.java | 4 +-
.../internal/HostComponentResourceProvider.java | 2 +-
.../internal/ServiceResourceProvider.java | 4 +-
.../ambari/server/orm/entities/StackEntity.java | 10 +++-
.../server/topology/HostOfferResponse.java | 46 ++++++++++++------
.../ambari/server/topology/HostRequest.java | 18 +++----
.../ambari/server/topology/LogicalRequest.java | 13 +++---
.../ambari/server/topology/TopologyManager.java | 49 ++++++++++++++------
.../server/configuration/ConfigurationTest.java | 6 +--
11 files changed, 143 insertions(+), 66 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ambari/blob/5dea886e/ambari-server/src/main/java/org/apache/ambari/server/configuration/Configuration.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/configuration/Configuration.java b/ambari-server/src/main/java/org/apache/ambari/server/configuration/Configuration.java
index 9d2243b..15f186b 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/configuration/Configuration.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/configuration/Configuration.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -100,7 +100,6 @@ import com.google.gson.JsonPrimitive;
import com.google.inject.Inject;
import com.google.inject.Singleton;
-
/**
* The {@link Configuration} class is used to read from the
* {{ambari.properties}} file and manage/expose the configuration properties.
@@ -2481,6 +2480,22 @@ public class Configuration {
public static final ConfigurationProperty<Long> LOG4JMONITOR_DELAY = new ConfigurationProperty<>(
"log4j.monitor.delay", TimeUnit.MINUTES.toMillis(5));
+ /**
+ * Indicates whether parallel topology task creation is enabled for blueprint cluster provisioning.
+ * Defaults to <code>false</code>.
+ * @see #TOPOLOGY_TASK_PARALLEL_CREATION_THREAD_COUNT
+ */
+ @Markdown(description = "Indicates whether parallel topology task creation is enabled")
+ public static final ConfigurationProperty<Boolean> TOPOLOGY_TASK_PARALLEL_CREATION_ENABLED = new ConfigurationProperty<>("topology.task.creation.parallel", Boolean.FALSE);
+
+ /**
+ * The number of threads to use for parallel topology task creation in blueprint cluster provisioning if enabled.
+ * Defaults to 10.
+ * @see #TOPOLOGY_TASK_PARALLEL_CREATION_ENABLED
+ */
+ @Markdown(description = "The number of threads to use for parallel topology task creation if enabled")
+ public static final ConfigurationProperty<Integer> TOPOLOGY_TASK_PARALLEL_CREATION_THREAD_COUNT = new ConfigurationProperty<>("topology.task.creation.parallel.threads", 10);
+
private static final Logger LOG = LoggerFactory.getLogger(
Configuration.class);
@@ -5168,6 +5183,24 @@ public class Configuration {
}
/**
+ * @return the number of threads to use for parallel topology task creation if enabled
+ */
+ public int getParallelTopologyTaskCreationThreadPoolSize() {
+ try {
+ return Integer.parseInt(getProperty(TOPOLOGY_TASK_PARALLEL_CREATION_THREAD_COUNT));
+ } catch (NumberFormatException e) {
+ return TOPOLOGY_TASK_PARALLEL_CREATION_THREAD_COUNT.getDefaultValue();
+ }
+ }
+
+ /**
+ * @return true if parallel execution of task creation is enabled explicitly
+ */
+ public boolean isParallelTopologyTaskCreationEnabled() {
+ return Boolean.parseBoolean(getProperty(TOPOLOGY_TASK_PARALLEL_CREATION_ENABLED));
+ }
+
+ /**
* Generates a markdown table which includes:
* <ul>
* <li>Property key name</li>
http://git-wip-us.apache.org/repos/asf/ambari/blob/5dea886e/ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariManagementControllerImpl.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariManagementControllerImpl.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariManagementControllerImpl.java
index 09e49ef..b04fdd7 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariManagementControllerImpl.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariManagementControllerImpl.java
@@ -2164,16 +2164,16 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
commandParams.putAll(commandParamsInp);
}
- //Propogate HCFS service type info
- Iterator<Service> it = cluster.getServices().values().iterator();
- while(it.hasNext()) {
- ServiceInfo serviceInfoInstance = ambariMetaInfo.getService(stackId.getStackName(),stackId.getStackVersion(), it.next().getName());
- LOG.info("Iterating service type Instance in createHostAction:: " + serviceInfoInstance.getName());
- if(serviceInfoInstance.getServiceType() != null) {
- LOG.info("Adding service type info in createHostAction:: " + serviceInfoInstance.getServiceType());
- commandParams.put("dfs_type",serviceInfoInstance.getServiceType());
- break;
- }
+ // Propagate HCFS service type info
+ for (Service service : cluster.getServices().values()) {
+ ServiceInfo serviceInfoInstance = ambariMetaInfo.getService(stackId.getStackName(),stackId.getStackVersion(), service.getName());
+ LOG.debug("Iterating service type Instance in createHostAction: {}", serviceInfoInstance.getName());
+ String serviceType = serviceInfoInstance.getServiceType();
+ if (serviceType != null) {
+ LOG.info("Adding service type info in createHostAction: {}", serviceType);
+ commandParams.put("dfs_type", serviceType);
+ break;
+ }
}
boolean isInstallCommand = roleCommand.equals(RoleCommand.INSTALL);
http://git-wip-us.apache.org/repos/asf/ambari/blob/5dea886e/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ComponentResourceProvider.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ComponentResourceProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ComponentResourceProvider.java
index 241a48f..453c688 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ComponentResourceProvider.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ComponentResourceProvider.java
@@ -266,7 +266,7 @@ public class ComponentResourceProvider extends AbstractControllerResourceProvide
}
// Create the components for the given requests.
- public synchronized void createComponents(Set<ServiceComponentRequest> requests)
+ public void createComponents(Set<ServiceComponentRequest> requests)
throws AmbariException, AuthorizationException {
if (requests.isEmpty()) {
@@ -473,7 +473,7 @@ public class ComponentResourceProvider extends AbstractControllerResourceProvide
}
// Update the components for the given requests.
- protected synchronized RequestStatusResponse updateComponents(Set<ServiceComponentRequest> requests,
+ protected RequestStatusResponse updateComponents(Set<ServiceComponentRequest> requests,
Map<String, String> requestProperties,
boolean runSmokeTest)
throws AmbariException, AuthorizationException {
http://git-wip-us.apache.org/repos/asf/ambari/blob/5dea886e/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/HostComponentResourceProvider.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/HostComponentResourceProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/HostComponentResourceProvider.java
index 87eb266..c8ec08b 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/HostComponentResourceProvider.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/HostComponentResourceProvider.java
@@ -474,7 +474,7 @@ public class HostComponentResourceProvider extends AbstractControllerResourcePro
//todo: This was moved from AmbariManagementController and needs a lot of refactoring.
//todo: Look into using the predicate instead of Set<ServiceComponentHostRequest>
//todo: change to private access when all AMC tests have been moved.
- protected synchronized RequestStageContainer updateHostComponents(RequestStageContainer stages,
+ protected RequestStageContainer updateHostComponents(RequestStageContainer stages,
Set<ServiceComponentHostRequest> requests,
Map<String, String> requestProperties,
boolean runSmokeTest) throws AmbariException, AuthorizationException {
http://git-wip-us.apache.org/repos/asf/ambari/blob/5dea886e/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ServiceResourceProvider.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ServiceResourceProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ServiceResourceProvider.java
index a08d153..0d5c174 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ServiceResourceProvider.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ServiceResourceProvider.java
@@ -343,7 +343,7 @@ public class ServiceResourceProvider extends AbstractControllerResourceProvider
}
// Create services from the given request.
- public synchronized void createServices(Set<ServiceRequest> requests)
+ public void createServices(Set<ServiceRequest> requests)
throws AmbariException, AuthorizationException {
if (requests.isEmpty()) {
@@ -461,7 +461,7 @@ public class ServiceResourceProvider extends AbstractControllerResourceProvider
}
// Update services based on the given requests.
- protected synchronized RequestStageContainer updateServices(RequestStageContainer requestStages, Set<ServiceRequest> requests,
+ protected RequestStageContainer updateServices(RequestStageContainer requestStages, Set<ServiceRequest> requests,
Map<String, String> requestProperties, boolean runSmokeTest,
boolean reconfigureClients, boolean startDependencies) throws AmbariException, AuthorizationException {
http://git-wip-us.apache.org/repos/asf/ambari/blob/5dea886e/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/StackEntity.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/StackEntity.java b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/StackEntity.java
index c425969..2cfe07c 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/StackEntity.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/orm/entities/StackEntity.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -24,6 +24,7 @@ import javax.persistence.GenerationType;
import javax.persistence.Id;
import javax.persistence.NamedQueries;
import javax.persistence.NamedQuery;
+import javax.persistence.QueryHint;
import javax.persistence.Table;
import javax.persistence.TableGenerator;
import javax.persistence.UniqueConstraint;
@@ -39,7 +40,12 @@ import javax.persistence.UniqueConstraint;
@TableGenerator(name = "stack_id_generator", table = "ambari_sequences", pkColumnName = "sequence_name", valueColumnName = "sequence_value", pkColumnValue = "stack_id_seq", initialValue = 0)
@NamedQueries({
@NamedQuery(name = "StackEntity.findAll", query = "SELECT stack FROM StackEntity stack"),
- @NamedQuery(name = "StackEntity.findByNameAndVersion", query = "SELECT stack FROM StackEntity stack WHERE stack.stackName = :stackName AND stack.stackVersion = :stackVersion") })
+ @NamedQuery(name = "StackEntity.findByNameAndVersion", query = "SELECT stack FROM StackEntity stack WHERE stack.stackName = :stackName AND stack.stackVersion = :stackVersion",
+ hints = {
+ @QueryHint(name = "eclipselink.query-results-cache", value = "true"),
+ @QueryHint(name = "eclipselink.query-results-cache.size", value = "100")
+ })
+})
public class StackEntity {
@Id
http://git-wip-us.apache.org/repos/asf/ambari/blob/5dea886e/ambari-server/src/main/java/org/apache/ambari/server/topology/HostOfferResponse.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/topology/HostOfferResponse.java b/ambari-server/src/main/java/org/apache/ambari/server/topology/HostOfferResponse.java
index 2932581..495aea6 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/topology/HostOfferResponse.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/topology/HostOfferResponse.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -19,30 +19,37 @@
package org.apache.ambari.server.topology;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.util.List;
+import java.util.concurrent.Executor;
/**
* Response to a host offer.
*/
-public class HostOfferResponse {
+final class HostOfferResponse {
+
public enum Answer {ACCEPTED, DECLINED_PREDICATE, DECLINED_DONE}
+ private static final Logger LOG = LoggerFactory.getLogger(HostOfferResponse.class);
+ static final HostOfferResponse DECLINED_DUE_TO_PREDICATE = new HostOfferResponse(Answer.DECLINED_PREDICATE);
+ static final HostOfferResponse DECLINED_DUE_TO_DONE = new HostOfferResponse(Answer.DECLINED_DONE);
+
private final Answer answer;
private final String hostGroupName;
private final long hostRequestId;
private final List<TopologyTask> tasks;
- public HostOfferResponse(Answer answer) {
- if (answer == Answer.ACCEPTED) {
- throw new IllegalArgumentException("For accepted response, hostgroup name and tasks must be set");
- }
- this.answer = answer;
- this.hostRequestId = -1;
- this.hostGroupName = null;
- this.tasks = null;
+ static HostOfferResponse createAcceptedResponse(long hostRequestId, String hostGroupName, List<TopologyTask> tasks) {
+ return new HostOfferResponse(Answer.ACCEPTED, hostRequestId, hostGroupName, tasks);
+ }
+
+ private HostOfferResponse(Answer answer) {
+ this(answer, -1, null, null);
}
- public HostOfferResponse(Answer answer, long hostRequestId, String hostGroupName, List<TopologyTask> tasks) {
+ private HostOfferResponse(Answer answer, long hostRequestId, String hostGroupName, List<TopologyTask> tasks) {
this.answer = answer;
this.hostRequestId = hostRequestId;
this.hostGroupName = hostGroupName;
@@ -63,7 +70,20 @@ public class HostOfferResponse {
return hostGroupName;
}
- public List<TopologyTask> getTasks() {
- return tasks;
+ void executeTasks(Executor executor, final String hostName, final ClusterTopology topology, final AmbariContext ambariContext) {
+ if (answer != Answer.ACCEPTED) {
+ LOG.warn("Attempted to execute tasks for declined host offer", answer);
+ } else {
+ executor.execute(new Runnable() {
+ @Override
+ public void run() {
+ for (TopologyTask task : tasks) {
+ LOG.info("Running task for accepted host offer for hostname = {}, task = {}", hostName, task.getType());
+ task.init(topology, ambariContext);
+ task.run();
+ }
+ }
+ });
+ }
}
}
http://git-wip-us.apache.org/repos/asf/ambari/blob/5dea886e/ambari-server/src/main/java/org/apache/ambari/server/topology/HostRequest.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/topology/HostRequest.java b/ambari-server/src/main/java/org/apache/ambari/server/topology/HostRequest.java
index 6a65b48..a18999b 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/topology/HostRequest.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/topology/HostRequest.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -134,15 +134,15 @@ public class HostRequest implements Comparable<HostRequest> {
//todo: synchronization
public synchronized HostOfferResponse offer(HostImpl host) {
if (!isOutstanding) {
- return new HostOfferResponse(HostOfferResponse.Answer.DECLINED_DONE);
+ return HostOfferResponse.DECLINED_DUE_TO_DONE;
}
if (matchesHost(host)) {
isOutstanding = false;
hostname = host.getHostName();
setHostOnTasks(host);
- return new HostOfferResponse(HostOfferResponse.Answer.ACCEPTED, id, hostGroup.getName(), topologyTasks);
+ return HostOfferResponse.createAcceptedResponse(id, hostGroup.getName(), topologyTasks);
} else {
- return new HostOfferResponse(HostOfferResponse.Answer.DECLINED_PREDICATE);
+ return HostOfferResponse.DECLINED_DUE_TO_PREDICATE;
}
}
@@ -466,6 +466,7 @@ public class HostRequest implements Comparable<HostRequest> {
@Override
public void run() {
+ LOG.info("HostRequest: Executing RESOURCE_CREATION task for host: {}", hostname);
HostGroup group = topology.getBlueprint().getHostGroup(getHostgroupName());
Map<String, Collection<String>> serviceComponents = new HashMap<String, Collection<String>>();
for (String service : group.getServices()) {
@@ -492,6 +493,7 @@ public class HostRequest implements Comparable<HostRequest> {
@Override
public void run() {
+ LOG.info("HostRequest: Executing CONFIGURE task for host: {}", hostname);
ambariContext.registerHostWithConfigGroup(getHostName(), clusterTopology, getHostgroupName());
}
}
@@ -517,7 +519,7 @@ public class HostRequest implements Comparable<HostRequest> {
@Override
public void run() {
- LOG.info("HostRequest.InstallHostTask: Executing INSTALL task for host: " + hostname);
+ LOG.info("HostRequest: Executing INSTALL task for host: {}", hostname);
boolean skipInstallTaskCreate = topology.getProvisionAction().equals(ProvisionAction.START_ONLY);
RequestStatusResponse response = clusterTopology.installHost(hostname, skipInstallTaskCreate, skipFailure);
// map logical install tasks to physical install tasks
@@ -543,7 +545,7 @@ public class HostRequest implements Comparable<HostRequest> {
}
}
- LOG.info("HostRequest.InstallHostTask: Exiting INSTALL task for host: " + hostname);
+ LOG.info("HostRequest: Exiting INSTALL task for host: {}", hostname);
}
}
@@ -568,7 +570,7 @@ public class HostRequest implements Comparable<HostRequest> {
@Override
public void run() {
- LOG.info("HostRequest.StartHostTask: Executing START task for host: " + hostname);
+ LOG.info("HostRequest: Executing START task for host: {}", hostname);
RequestStatusResponse response = clusterTopology.startHost(hostname, skipFailure);
// map logical install tasks to physical install tasks
List<ShortTaskStatus> underlyingTasks = response.getTasks();
@@ -591,7 +593,7 @@ public class HostRequest implements Comparable<HostRequest> {
}
}
- LOG.info("HostRequest.StartHostTask: Exiting START task for host: " + hostname);
+ LOG.info("HostRequest: Exiting START task for host: {}", hostname);
}
}
http://git-wip-us.apache.org/repos/asf/ambari/blob/5dea886e/ambari-server/src/main/java/org/apache/ambari/server/topology/LogicalRequest.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/topology/LogicalRequest.java b/ambari-server/src/main/java/org/apache/ambari/server/topology/LogicalRequest.java
index 3aaf589..0039e35 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/topology/LogicalRequest.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/topology/LogicalRequest.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -121,7 +121,7 @@ public class LogicalRequest extends Request {
//todo: prioritization of master host requests
Iterator<HostRequest> hostRequestIterator = outstandingHostRequests.iterator();
while (hostRequestIterator.hasNext()) {
- LOG.info("LogicalRequest.offer: attempting to match a request to a request for a reserved host to hostname = {}", host.getHostName());
+ LOG.info("LogicalRequest.offer: attempting to match a request to a request for a non-reserved host to hostname = {}", host.getHostName());
HostOfferResponse response = hostRequestIterator.next().offer(host);
switch (response.getAnswer()) {
case ACCEPTED:
@@ -132,23 +132,22 @@ public class LogicalRequest extends Request {
//todo: should have been done on ACCEPT
hostRequestIterator.remove();
LOG.info("LogicalRequest.offer: host request returned DECLINED_DONE for hostname = {}, host request has been removed from list", host.getHostName());
+ break;
case DECLINED_PREDICATE:
LOG.info("LogicalRequest.offer: host request returned DECLINED_PREDICATE for hostname = {}", host.getHostName());
predicateRejected = true;
+ break;
}
}
LOG.info("LogicalRequest.offer: outstandingHost request list size = " + outstandingHostRequests.size());
}
-
-
-
// if at least one outstanding host request rejected for predicate or we have an outstanding request
// with a reserved host decline due to predicate, otherwise decline due to all hosts being resolved
return predicateRejected || ! requestsWithReservedHosts.isEmpty() ?
- new HostOfferResponse(HostOfferResponse.Answer.DECLINED_PREDICATE) :
- new HostOfferResponse(HostOfferResponse.Answer.DECLINED_DONE);
+ HostOfferResponse.DECLINED_DUE_TO_PREDICATE :
+ HostOfferResponse.DECLINED_DUE_TO_DONE;
}
@Override
http://git-wip-us.apache.org/repos/asf/ambari/blob/5dea886e/ambari-server/src/main/java/org/apache/ambari/server/topology/TopologyManager.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/topology/TopologyManager.java b/ambari-server/src/main/java/org/apache/ambari/server/topology/TopologyManager.java
index bba0325..341633e 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/topology/TopologyManager.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/topology/TopologyManager.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -29,6 +29,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
+import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -38,6 +39,7 @@ import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.actionmanager.HostRoleCommand;
import org.apache.ambari.server.actionmanager.HostRoleStatus;
import org.apache.ambari.server.api.services.stackadvisor.StackAdvisorBlueprintProcessor;
+import org.apache.ambari.server.configuration.Configuration;
import org.apache.ambari.server.controller.AmbariServer;
import org.apache.ambari.server.controller.RequestStatusResponse;
import org.apache.ambari.server.controller.ShortTaskStatus;
@@ -86,7 +88,9 @@ public class TopologyManager {
private static final String CLUSTER_CONFIG_TASK_MAX_TIME_IN_MILLIS_PROPERTY_NAME = "cluster_configure_task_timeout";
private PersistedState persistedState;
- private ExecutorService executor = Executors.newSingleThreadExecutor();
+ private final ExecutorService executor = Executors.newSingleThreadExecutor();
+ private final Executor taskExecutor; // executes TopologyTasks
+ private final boolean parallelTaskCreationEnabled;
private Collection<String> hostsToIgnore = new HashSet<String>();
private final List<HostImpl> availableHosts = new LinkedList<HostImpl>();
private final Map<String, LogicalRequest> reservedHosts = new HashMap<String, LogicalRequest>();
@@ -127,8 +131,18 @@ public class TopologyManager {
*/
private Map<Long, Boolean> clusterProvisionWithBlueprintCreationFinished = new HashMap<>();
- public TopologyManager(){
+ public TopologyManager() {
+ parallelTaskCreationEnabled = false;
+ taskExecutor = executor;
+ }
+ @Inject
+ public TopologyManager(Configuration configuration) {
+ int threadPoolSize = configuration.getParallelTopologyTaskCreationThreadPoolSize();
+ parallelTaskCreationEnabled = configuration.isParallelTopologyTaskCreationEnabled() && threadPoolSize > 1;
+ taskExecutor = parallelTaskCreationEnabled
+ ? Executors.newFixedThreadPool(threadPoolSize)
+ : executor;
}
@Inject
@@ -691,7 +705,7 @@ public class TopologyManager {
return logicalRequest;
}
- private void processAcceptedHostOffer(ClusterTopology topology, final HostOfferResponse response, final HostImpl host) {
+ private void processAcceptedHostOffer(final ClusterTopology topology, final HostOfferResponse response, final HostImpl host) {
final String hostName = host.getHostName();
try {
topology.addHostToTopology(response.getHostGroupName(), hostName);
@@ -722,19 +736,24 @@ public class TopologyManager {
throw new RuntimeException(e);
}
-
- LOG.info("TopologyManager.processAcceptedHostOffer: about to execute tasks for host = {}",
- hostName);
-
- for (TopologyTask task : response.getTasks()) {
- LOG.info("Processing accepted host offer for {} which responded {} and task {}",
- hostName, response.getAnswer(), task.getType());
-
- task.init(topology, ambariContext);
- executor.execute(task);
+ LOG.info("TopologyManager.processAcceptedHostOffer: queue tasks for host = {} which responded {}", hostName, response.getAnswer());
+ if (parallelTaskCreationEnabled) {
+ executor.execute(new Runnable() { // do not start until cluster config done
+ @Override
+ public void run() {
+ queueHostTasks(topology, response, hostName);
+ }
+ });
+ } else {
+ queueHostTasks(topology, response, hostName);
}
}
+ private void queueHostTasks(ClusterTopology topology, HostOfferResponse response, String hostName) {
+ LOG.info("TopologyManager.processAcceptedHostOffer: queueing tasks for host = {}", hostName);
+ response.executeTasks(taskExecutor, hostName, topology, ambariContext);
+ }
+
private void updateHostWithRackInfo(ClusterTopology topology, HostOfferResponse response, HostImpl host) {
// the rack info from the cluster creation template
String rackInfoFromTemplate = topology.getHostGroupInfo().get(response.getHostGroupName()).getHostRackInfo().get
@@ -878,7 +897,7 @@ public class TopologyManager {
}
ConfigureClusterTask configureClusterTask = new ConfigureClusterTask(topology, configurationRequest);
- AsyncCallableService<Boolean> asyncCallableService = new AsyncCallableService(configureClusterTask, timeout, delay,
+ AsyncCallableService<Boolean> asyncCallableService = new AsyncCallableService<>(configureClusterTask, timeout, delay,
Executors.newScheduledThreadPool(1));
executor.submit(asyncCallableService);
http://git-wip-us.apache.org/repos/asf/ambari/blob/5dea886e/ambari-server/src/test/java/org/apache/ambari/server/configuration/ConfigurationTest.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/test/java/org/apache/ambari/server/configuration/ConfigurationTest.java b/ambari-server/src/test/java/org/apache/ambari/server/configuration/ConfigurationTest.java
index f90cf76..b0bcc58 100644
--- a/ambari-server/src/test/java/org/apache/ambari/server/configuration/ConfigurationTest.java
+++ b/ambari-server/src/test/java/org/apache/ambari/server/configuration/ConfigurationTest.java
@@ -1039,8 +1039,6 @@ public class ConfigurationTest {
/**
* Tests that every {@link ConfigurationProperty} field in
* {@link Configuration} has a property {@link Markdown} annotation.
- *
- * @throws Exception
*/
@Test
public void testAllPropertiesHaveMarkdownDescriptions() throws Exception {
@@ -1052,9 +1050,9 @@ public class ConfigurationTest {
ConfigurationProperty<?> configurationProperty = (ConfigurationProperty<?>) field.get(null);
Markdown markdown = field.getAnnotation(Markdown.class);
- if( null == markdown ){
+ if (null == markdown) {
ConfigurationMarkdown configMarkdown = field.getAnnotation(ConfigurationMarkdown.class);
- markdown = configMarkdown.markdown();
+ markdown = configMarkdown != null ? configMarkdown.markdown() : null;
}
Assert.assertNotNull("The configuration property " + configurationProperty.getKey()