AMBARI-15226 - The stack advisor should be pushed down to the services (Tim Thorpe via jluniya)

ambari-server/src/main/assemblies/server.xml

@@ -308,6 +308,11 @@
       <source>src/main/resources/stacks/stack_advisor.py</source>
       <outputDirectory>/var/lib/ambari-server/resources/stacks</outputDirectory>
     </file>
+    <file>
+      <fileMode>755</fileMode>
+      <source>src/main/resources/stacks/service_advisor.py</source>
+      <outputDirectory>/var/lib/ambari-server/resources/stacks</outputDirectory>
+    </file>
     <file>
       <fileMode>755</fileMode>
       <source>src/main/python/bootstrap.py</source>

ambari-server/src/main/java/org/apache/ambari/server/api/services/AmbariMetaInfo.java

@@ -104,6 +104,7 @@ public class AmbariMetaInfo {
   public static final String RCO_FILE_NAME = "role_command_order.json";
   public static final String SERVICE_METRIC_FILE_NAME = "metrics.json";
   public static final String SERVICE_ALERT_FILE_NAME = "alerts.json";
+  public static final String SERVICE_ADVISOR_FILE_NAME = "service_advisor.py";
 
   /**
    * The filename name for a Kerberos descriptor file at either the stack or service level

ambari-server/src/main/java/org/apache/ambari/server/api/services/stackadvisor/commands/StackAdvisorCommand.java

@@ -44,6 +44,7 @@ import org.apache.ambari.server.api.services.stackadvisor.StackAdvisorRequest;
 import org.apache.ambari.server.api.services.stackadvisor.StackAdvisorResponse;
 import org.apache.ambari.server.api.services.stackadvisor.StackAdvisorRunner;
 import org.apache.ambari.server.controller.spi.Resource;
+import org.apache.ambari.server.state.ServiceInfo;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.io.FileUtils;
 import org.apache.commons.lang.StringUtils;
@@ -83,6 +84,7 @@ public abstract class StackAdvisorCommand<T extends StackAdvisorResponse> extend
   private static final String SERVICES_PROPERTY = "services";
   private static final String SERVICES_COMPONENTS_PROPERTY = "components";
   private static final String CONFIG_GROUPS_PROPERTY = "config-groups";
+  private static final String STACK_SERVICES_PROPERTY = "StackServices";
   private static final String COMPONENT_INFO_PROPERTY = "StackServiceComponents";
   private static final String COMPONENT_NAME_PROPERTY = "component_name";
   private static final String COMPONENT_HOSTNAMES_PROPERTY = "hostnames";
@@ -135,7 +137,7 @@ public abstract class StackAdvisorCommand<T extends StackAdvisorResponse> extend
   /**
    * Name with the result JSON, e.g. "component-layout.json" or
    * "validations.json" .
-   * 
+   *
    * @return the file name
    */
   protected abstract String getResultFileName();
@@ -148,6 +150,7 @@ public abstract class StackAdvisorCommand<T extends StackAdvisorResponse> extend
 
       populateStackHierarchy(root);
       populateComponentHostsMap(root, request.getComponentHostsMap());
+      populateServiceAdvisors(root);
       populateConfigurations(root, request);
       populateConfigGroups(root, request);
       populateAmbariServerInfo(root);
@@ -241,6 +244,31 @@ public abstract class StackAdvisorCommand<T extends StackAdvisorResponse> extend
     }
   }
 
+  private void populateServiceAdvisors(ObjectNode root) {
+    ArrayNode services = (ArrayNode) root.get(SERVICES_PROPERTY);
+    Iterator<JsonNode> servicesIter = services.getElements();
+
+    ObjectNode version = (ObjectNode) root.get("Versions");
+    String stackName = version.get("stack_name").asText();
+    String stackVersion = version.get("stack_version").asText();
+
+    while (servicesIter.hasNext()) {
+      JsonNode service = servicesIter.next();
+      ObjectNode serviceVersion = (ObjectNode) service.get(STACK_SERVICES_PROPERTY);
+      String serviceName = serviceVersion.get("service_name").getTextValue();
+      try {
+        ServiceInfo serviceInfo = metaInfo.getService(stackName, stackVersion, serviceName);
+        if (serviceInfo.getAdvisorFile() != null) {
+          serviceVersion.put("advisor_name", serviceInfo.getAdvisorName());
+          serviceVersion.put("advisor_path", serviceInfo.getAdvisorFile().getAbsolutePath());
+        }
+      }
+      catch (Exception e) {
+        LOG.error("Error adding service advisor information to services.json", e);
+      }
+    }
+  }
+
   public synchronized T invoke(StackAdvisorRequest request) throws StackAdvisorException {
     validate(request);
     String hostsJSON = getHostsInformation(request);

ambari-server/src/main/java/org/apache/ambari/server/controller/StackServiceResponse.java

@@ -113,7 +113,7 @@ public class StackServiceResponse {
   public void setServiceName(String serviceName) {
     this.serviceName = serviceName;
   }
-  
+
   public String getServiceType() {
 	return serviceType;
   }
@@ -161,7 +161,7 @@ public String getServiceDisplayName() {
   public Set<String> getExcludedConfigTypes() {
     return excludedConfigTypes;
   }
-  
+
   public List<String> getRequiredServices() {
     return requiredServices;
   }

ambari-server/src/main/java/org/apache/ambari/server/controller/internal/StackServiceResourceProvider.java

@@ -48,9 +48,9 @@ public class StackServiceResourceProvider extends ReadOnlyResourceProvider {
 
   protected static final String SERVICE_NAME_PROPERTY_ID = PropertyHelper.getPropertyId(
       "StackServices", "service_name");
-  
+
   protected static final String SERVICE_TYPE_PROPERTY_ID = PropertyHelper.getPropertyId(
-		  "StackServices", "service_type"); 
+		  "StackServices", "service_type");
 
   public static final String STACK_NAME_PROPERTY_ID = PropertyHelper.getPropertyId(
       "StackServices", "stack_name");
@@ -72,7 +72,7 @@ public class StackServiceResourceProvider extends ReadOnlyResourceProvider {
 
   private static final String CONFIG_TYPES = PropertyHelper.getPropertyId(
       "StackServices", "config_types");
-  
+
   private static final String REQUIRED_SERVICES_ID = PropertyHelper.getPropertyId(
       "StackServices", "required_services");
 
@@ -144,9 +144,9 @@ public class StackServiceResourceProvider extends ReadOnlyResourceProvider {
 
       setResourceProperty(resource, SERVICE_NAME_PROPERTY_ID,
           response.getServiceName(), requestedIds);
-      
+
       setResourceProperty(resource, SERVICE_TYPE_PROPERTY_ID,
-    		  response.getServiceType(), requestedIds);
+		  response.getServiceType(), requestedIds);
 
     setResourceProperty(resource, STACK_VERSION_PROPERTY_ID,
         response.getStackVersion(), requestedIds);

ambari-server/src/main/java/org/apache/ambari/server/stack/CommonServiceDirectory.java

@@ -43,6 +43,23 @@ public class CommonServiceDirectory extends ServiceDirectory {
     super(servicePath);
   }
 
+  @Override
+  /**
+   * Obtain the advisor name.
+   *
+   * @return advisor name
+   */
+  public String getAdvisorName(String serviceName) {
+    if (getAdvisorFile() == null || serviceName == null)
+      return null;
+
+    File serviceVersionDir = new File(getAbsolutePath());
+    String serviceVersion = serviceVersionDir.getName().replaceAll("\\.", "");
+
+    String advisorName = serviceName + serviceVersion + "ServiceAdvisor";
+    return advisorName;
+  }
+
   @Override
   /**
    * Parse common service directory

ambari-server/src/main/java/org/apache/ambari/server/stack/ServiceDirectory.java

@@ -39,6 +39,11 @@ public abstract class ServiceDirectory extends StackDefinitionDirectory {
    */
   private Map<String, File> metricsFileMap = new HashMap<String, File>();
 
+  /**
+   * advisor file
+   */
+  private File advisorFile;
+
   /**
    * alerts file
    */
@@ -125,6 +130,10 @@ public abstract class ServiceDirectory extends StackDefinitionDirectory {
       }
     }
 
+    File advFile = new File(directory.getAbsolutePath()
+        + File.separator + AmbariMetaInfo.SERVICE_ADVISOR_FILE_NAME);
+    advisorFile = advFile.exists() ? advFile : null;
+
     File themeFile = new File(directory.getAbsolutePath() + File.separator + AmbariMetaInfo.SERVICE_THEME_FILE_NAME);
     this.themeFile = themeFile.exists() ? themeFile : null;
   }
@@ -147,6 +156,22 @@ public abstract class ServiceDirectory extends StackDefinitionDirectory {
     return metricsFileMap.get(serviceName);
   }
 
+  /**
+   * Obtain the advisor file.
+   *
+   * @return advisor file
+   */
+  public File getAdvisorFile() {
+    return advisorFile;
+  }
+
+  /**
+   * Obtain the advisor name.
+   *
+   * @return advisor name
+   */
+  public abstract String getAdvisorName(String serviceName);
+
   /**
    * Obtain the alerts file.
    *

ambari-server/src/main/java/org/apache/ambari/server/stack/ServiceModule.java

@@ -132,6 +132,8 @@ public class ServiceModule extends BaseModule<ServiceModule, ServiceInfo> implem
     serviceInfo.setWidgetsDescriptorFile(serviceDirectory.getWidgetsDescriptorFile(serviceInfo.getName()));
     serviceInfo.setSchemaVersion(AmbariMetaInfo.SCHEMA_VERSION_2);
     serviceInfo.setServicePackageFolder(serviceDirectory.getPackageDir());
+    serviceInfo.setAdvisorFile(serviceDirectory.getAdvisorFile());
+    serviceInfo.setAdvisorName(serviceDirectory.getAdvisorName(serviceInfo.getName()));
 
     populateComponentModules();
     populateConfigurationModules();
@@ -174,7 +176,7 @@ public class ServiceModule extends BaseModule<ServiceModule, ServiceInfo> implem
     }
 
     ServiceInfo parent = parentModule.getModuleInfo();
-    
+
     if (serviceInfo.getComment() == null) {
       serviceInfo.setComment(parent.getComment());
     }
@@ -225,6 +227,12 @@ public class ServiceModule extends BaseModule<ServiceModule, ServiceInfo> implem
     if (serviceInfo.getWidgetsDescriptorFile() == null) {
       serviceInfo.setWidgetsDescriptorFile(parent.getWidgetsDescriptorFile());
     }
+    if (serviceInfo.getAdvisorFile() == null) {
+      serviceInfo.setAdvisorFile(parent.getAdvisorFile());
+    }
+    if (serviceInfo.getAdvisorName() == null) {
+      serviceInfo.setAdvisorName(parent.getAdvisorName());
+    }
 
     mergeCustomCommands(parent.getCustomCommands(), serviceInfo.getCustomCommands());
     mergeConfigDependencies(parent);
@@ -359,7 +367,7 @@ public class ServiceModule extends BaseModule<ServiceModule, ServiceInfo> implem
               setErrors(config.getErrors());
               setErrors(info.getErrors());
             }
-          }          
+          }
           serviceInfo.getProperties().addAll(info.getProperties());
           serviceInfo.setTypeAttributes(config.getConfigType(), info.getAttributes());
           configurationModules.put(config.getConfigType(), config);
@@ -582,9 +590,9 @@ public class ServiceModule extends BaseModule<ServiceModule, ServiceInfo> implem
   public void setValid(boolean valid) {
     this.valid = valid;
   }
-  
+
   private Set<String> errorSet = new HashSet<String>();
-  
+
   @Override
   public void setErrors(String error) {
     errorSet.add(error);
@@ -594,7 +602,7 @@ public class ServiceModule extends BaseModule<ServiceModule, ServiceInfo> implem
   public Collection getErrors() {
     return errorSet;
   }
-  
+
   @Override
   public void setErrors(Collection error) {
     this.errorSet.addAll(error);

ambari-server/src/main/java/org/apache/ambari/server/stack/StackServiceDirectory.java

@@ -45,6 +45,26 @@ public class StackServiceDirectory extends ServiceDirectory {
     super(servicePath);
   }
 
+  @Override
+  /**
+   * Obtain the advisor name.
+   *
+   * @return advisor name
+   */
+  public String getAdvisorName(String serviceName) {
+    if (getAdvisorFile() == null || serviceName == null)
+      return null;
+
+    File serviceDir = new File(getAbsolutePath());
+    File stackVersionDir = serviceDir.getParentFile().getParentFile();
+    File stackDir = stackVersionDir.getParentFile();
+
+    String stackName = stackDir.getName();
+    String versionString = stackVersionDir.getName().replaceAll("\\.", "");
+
+    return stackName + versionString + serviceName + "ServiceAdvisor";
+  }
+
   @Override
   /**
    * Parse stack service directory.

ambari-server/src/main/java/org/apache/ambari/server/state/ServiceInfo.java

@@ -95,7 +95,7 @@ public class ServiceInfo implements Validable{
 
   @JsonIgnore
   private Boolean monitoringService;
-  
+
   @JsonIgnore
   @XmlElement(name = "restartRequiredAfterChange")
   private Boolean restartRequiredAfterChange;
@@ -129,7 +129,13 @@ public class ServiceInfo implements Validable{
 
   @XmlTransient
   private Map<String, Map<String, List<MetricDefinition>>> metrics = null;
-  
+
+  @XmlTransient
+  private File advisorFile = null;
+
+  @XmlTransient
+  private String advisorName = null;
+
   @XmlTransient
   private File alertsFile = null;
 
@@ -138,7 +144,7 @@ public class ServiceInfo implements Validable{
 
   @XmlTransient
   private File widgetsDescriptorFile = null;
-  
+
   @XmlTransient
   private boolean valid = true;
 
@@ -161,7 +167,7 @@ public class ServiceInfo implements Validable{
   }
 
   /**
-   * 
+   *
    * @param valid set validity flag
    */
   @Override
@@ -171,7 +177,7 @@ public class ServiceInfo implements Validable{
 
   @XmlTransient
   private Set<String> errorSet = new HashSet<String>();
-  
+
   @Override
   public void setErrors(String error) {
     errorSet.add(error);
@@ -180,8 +186,8 @@ public class ServiceInfo implements Validable{
   @Override
   public Collection getErrors() {
     return errorSet;
-  }   
-  
+  }
+
   @Override
   public void setErrors(Collection error) {
     this.errorSet.addAll(error);
@@ -193,7 +199,7 @@ public class ServiceInfo implements Validable{
   @XmlElementWrapper(name="osSpecifics")
   @XmlElements(@XmlElement(name="osSpecific"))
   private List<ServiceOsSpecific> serviceOsSpecifics;
-  
+
   @JsonIgnore
   @XmlElement(name="configuration-dir")
   private String configDir = AmbariMetaInfo.SERVICE_CONFIG_FOLDER_NAME;
@@ -241,7 +247,7 @@ public class ServiceInfo implements Validable{
   @XmlElementWrapper(name="customCommands")
   @XmlElements(@XmlElement(name="customCommand"))
   private List<CustomCommandDefinition> customCommands;
-  
+
   @XmlElementWrapper(name="requiredServices")
   @XmlElement(name="service")
   private List<String> requiredServices = new ArrayList<String>();
@@ -286,7 +292,7 @@ public class ServiceInfo implements Validable{
   public void setDisplayName(String displayName) {
     this.displayName = displayName;
   }
-  
+
   public String getServiceType() {
 	return serviceType;
   }
@@ -381,13 +387,29 @@ public String getVersion() {
     return client;
   }
 
+  public File getAdvisorFile() {
+    return advisorFile;
+  }
+
+  public void setAdvisorFile(File advisorFile) {
+    this.advisorFile = advisorFile;
+  }
+
+  public String getAdvisorName() {
+    return advisorName;
+  }
+
+  public void setAdvisorName(String advisorName) {
+    this.advisorName = advisorName;
+  }
+
   @Override
   public String toString() {
     StringBuilder sb = new StringBuilder();
     sb.append("Service name:");
     sb.append(name);
     sb.append("\nService type:");
-    sb.append(serviceType); 
+    sb.append(serviceType);
     sb.append("\nversion:");
     sb.append(version);
     sb.append("\ncomment:");
@@ -475,7 +497,7 @@ public String getVersion() {
    * This can be used in determining if a property is stale.
 
    * @param type the config type
-   * @param keyNames the names of all the config keys for the given type 
+   * @param keyNames the names of all the config keys for the given type
    * @return <code>true</code> if the config is stale
    */
   public boolean hasDependencyAndPropertyFor(String type, Collection<String> keyNames) {
@@ -489,7 +511,7 @@ public String getVersion() {
       if (keys != null && keys.contains(staleCheck))
         return true;
     }
-    
+
     return false;
   }
 
@@ -612,7 +634,7 @@ public String getVersion() {
   public void setMetricsFile(File file) {
     metricsFile = file;
   }
-  
+
   /**
    * @return the metrics file, or <code>null</code> if none exists
    */
@@ -626,14 +648,14 @@ public String getVersion() {
   public Map<String, Map<String, List<MetricDefinition>>> getMetrics() {
     return metrics;
   }
-  
+
   /**
    * @param map the metrics for this service
    */
   public void setMetrics(Map<String, Map<String, List<MetricDefinition>>> map) {
     metrics = map;
   }
-  
+
   /**
    * @return the configuration directory name
    */

ambari-server/src/main/resources/common-services/HAWQ/2.0.0/service_advisor.py

@@ -0,0 +1,284 @@
+#!/usr/bin/env ambari-python-wrap
+"""
+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.
+"""
+import os
+import fnmatch
+import imp
+import re
+import socket
+import sys
+import traceback
+
+SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
+STACKS_DIR = os.path.join(SCRIPT_DIR, '../../../stacks/')
+PARENT_FILE = os.path.join(STACKS_DIR, 'service_advisor.py')
+
+try:
+  with open(PARENT_FILE, 'rb') as fp:
+    service_advisor = imp.load_module('service_advisor', fp, PARENT_FILE, ('.py', 'rb', imp.PY_SOURCE))
+except Exception as e:
+  traceback.print_exc()
+  print "Failed to load parent"
+
+class HAWQ200ServiceAdvisor(service_advisor.ServiceAdvisor):
+
+  def getHostsForMasterComponent(self, stackAdvisor, services, hosts, component, hostsList, hostsComponentsMap):
+    componentName = component["StackServiceComponents"]["component_name"]
+
+    if componentName == 'HAWQSTANDBY' and len(hostsList) == 1:
+      return []
+
+    return stackAdvisor.getHostsForMasterComponent(services, hosts, component, hostsList, hostsComponentsMap)
+
+  def isComponentNotPreferableOnAmbariServerHost(self, componentName):
+    return componentName in ('HAWQMASTER', 'HAWQSTANDBY')
+
+  def getComponentLayoutScheme(self, componentName):
+    if componentName == 'HAWQMASTER':
+      return {6: 2, 31: 1, "else": 5}
+
+    if componentName == 'HAWQSTANDBY':
+      return {6: 1, 31: 2, "else": 3}
+
+    return None
+
+  def colocateService(self, stackAdvisor, hostsComponentsMap, serviceComponents):
+    # colocate HAWQSEGMENT with DATANODE, if no hosts have been allocated for HAWQSEGMENT
+    hawqSegment = [component for component in serviceComponents if component["StackServiceComponents"]["component_name"] == "HAWQSEGMENT"][0]
+    if not stackAdvisor.isComponentHostsPopulated(hawqSegment):
+      for hostName in hostsComponentsMap.keys():
+        hostComponents = hostsComponentsMap[hostName]
+        if {"name": "DATANODE"} in hostComponents and {"name": "HAWQSEGMENT"} not in hostComponents:
+          hostsComponentsMap[hostName].append( { "name": "HAWQSEGMENT" } )
+        if {"name": "DATANODE"} not in hostComponents and {"name": "HAWQSEGMENT"} in hostComponents:
+          hostComponents.remove({"name": "HAWQSEGMENT"})
+
+  def getComponentLayoutValidations(self, stackAdvisor, services, hosts):
+    componentsListList = [service["components"] for service in services["services"]]
+    componentsList = [item["StackServiceComponents"] for sublist in componentsListList for item in sublist]
+    hostsList = [host["Hosts"]["host_name"] for host in hosts["items"]]
+    hostsCount = len(hostsList)
+
+    hawqMasterHosts = self.getHosts(componentsList, "HAWQMASTER")
+    hawqStandbyHosts = self.getHosts(componentsList, "HAWQSTANDBY")
+    hawqSegmentHosts = self.getHosts(componentsList, "HAWQSEGMENT")
+    datanodeHosts = self.getHosts(componentsList, "DATANODE")
+
+    items = []
+
+    # Generate WARNING if any HAWQSEGMENT is not colocated with a DATANODE
+    mismatchHosts = sorted(set(hawqSegmentHosts).symmetric_difference(set(datanodeHosts)))
+    if len(mismatchHosts) > 0:
+      hostsString = ', '.join(mismatchHosts)
+      message = "HAWQ Segment must be installed on all DataNodes. " \
+                  "The following {0} host(s) do not satisfy the colocation recommendation: {1}".format(len(mismatchHosts), hostsString)
+      items.append( { "type": 'host-component', "level": 'WARN', "message": message, "component-name": 'HAWQSEGMENT' } )
+
+    # single node case is not analyzed because HAWQ Standby Master will not be present in single node topology due to logic in createComponentLayoutRecommendations()
+    if len(hawqMasterHosts) == 1 and len(hawqStandbyHosts) == 1 and hawqMasterHosts == hawqStandbyHosts:
+      message = "HAWQ Master and HAWQ Standby Master cannot be deployed on the same host."
+      items.append( { "type": 'host-component', "level": 'ERROR', "message": message, "component-name": 'HAWQSTANDBY', "host": hawqStandbyHosts[0] } )
+
+    if len(hawqMasterHosts) ==  1 and hostsCount > 1 and stackAdvisor.isLocalHost(hawqMasterHosts[0]):
+      message = "The default Postgres port (5432) on the Ambari Server conflicts with the default HAWQ Masters port. " \
+                "If you are using port 5432 for Postgres, you must either deploy the HAWQ Master on a different host " \
+                "or configure a different port for the HAWQ Masters in the HAWQ Configuration page."
+      items.append( { "type": 'host-component', "level": 'WARN', "message": message, "component-name": 'HAWQMASTER', "host": hawqMasterHosts[0] } )
+
+    if len(hawqStandbyHosts) ==  1 and hostsCount > 1 and stackAdvisor.isLocalHost(hawqStandbyHosts[0]):
+      message = "The default Postgres port (5432) on the Ambari Server conflicts with the default HAWQ Masters port. " \
+                "If you are using port 5432 for Postgres, you must either deploy the HAWQ Standby Master on a different host " \
+                "or configure a different port for the HAWQ Masters in the HAWQ Configuration page."
+      items.append( { "type": 'host-component', "level": 'WARN', "message": message, "component-name": 'HAWQSTANDBY', "host": hawqStandbyHosts[0] } )
+
+    return items
+
+  def isHawqMasterComponentOnAmbariServer(self, stackAdvisor, services):
+    componentsListList = [service["components"] for service in services["services"]]
+    componentsList = [item for sublist in componentsListList for item in sublist]
+    hawqMasterComponentHosts = [hostname for component in componentsList if component["StackServiceComponents"]["component_name"] in ("HAWQMASTER", "HAWQSTANDBY") for hostname in component["StackServiceComponents"]["hostnames"]]
+    return any([stackAdvisor.isLocalHost(host) for host in hawqMasterComponentHosts])
+
+  def getServiceConfigurationRecommendations(self, stackAdvisor, configurations, clusterData, services, hosts):
+    putHdfsSiteProperty = self.putProperty(configurations, "hdfs-site", services)
+
+    # Set dfs.allow.truncate to true
+    putHdfsSiteProperty('dfs.allow.truncate', 'true')
+
+    if any(x in services["configurations"] for x in ["hawq-site", "hdfs-client"]):
+      componentsListList = [service["components"] for service in services["services"]]
+      componentsList = [item["StackServiceComponents"] for sublist in componentsListList for item in sublist]
+      servicesList = [service["StackServices"]["service_name"] for service in services["services"]]
+      numSegments = len(self.getHosts(componentsList, "HAWQSEGMENT"))
+
+    if "hawq-site" in services["configurations"]:
+      hawq_site = services["configurations"]["hawq-site"]["properties"]
+      putHawqSiteProperty = self.putProperty(configurations, "hawq-site", services)
+
+      # remove master port when master is colocated with Ambari server
+      if self.isHawqMasterComponentOnAmbariServer(stackAdvisor, services) and "hawq_master_address_port" in hawq_site:
+        putHawqSiteProperty('hawq_master_address_port', '')
+
+      # update query limits if segments are deployed
+      if numSegments and "default_hash_table_bucket_number" in hawq_site and "hawq_rm_nvseg_perquery_limit" in hawq_site:
+        factor_min = 1
+        factor_max = 6
+        limit = int(hawq_site["hawq_rm_nvseg_perquery_limit"])
+        factor = limit / numSegments
+        # if too many segments or default limit is too low --> stick with the limit
+        if factor < factor_min:
+          buckets = limit
+        # if the limit is large and results in factor > max --> limit factor to max
+        elif factor > factor_max:
+          buckets = factor_max * numSegments
+        else:
+          buckets = factor * numSegments
+        putHawqSiteProperty('default_hash_table_bucket_number', buckets)
+
+      # update YARN RM urls with the values from yarn-site if YARN is installed
+      if "YARN" in servicesList and "yarn-site" in services["configurations"]:
+        yarn_site = services["configurations"]["yarn-site"]["properties"]
+        for hs_prop, ys_prop in self.getHAWQYARNPropertyMapping().items():
+          if hs_prop in hawq_site and ys_prop in yarn_site:
+            putHawqSiteProperty(hs_prop, yarn_site[ys_prop])
+
+    # set output.replace-datanode-on-failure in HAWQ hdfs-client depending on the cluster size
+    if "hdfs-client" in services["configurations"]:
+      hdfs_client = services["configurations"]["hdfs-client"]["properties"]
+      if "output.replace-datanode-on-failure" in hdfs_client:
+        propertyValue = "true" if numSegments > 3 else "false"
+        putHdfsClientProperty = self.putProperty(configurations, "hdfs-client", services)
+        putHdfsClientProperty("output.replace-datanode-on-failure", propertyValue)
+
+  def getHAWQYARNPropertyMapping(self):
+    return { "hawq_rm_yarn_address": "yarn.resourcemanager.address", "hawq_rm_yarn_scheduler_address": "yarn.resourcemanager.scheduler.address" }
+
+  def getConfigurationsValidationItems(self, stackAdvisor, configurations, recommendedDefaults, services, hosts):
+    siteName = "hawq-site"
+    method = self.validateHAWQSiteConfigurations
+    items = self.validateConfigurationsForSite(stackAdvisor, configurations, recommendedDefaults, services, hosts, siteName, method)
+
+    siteName = "hdfs-client"
+    method = self.validateHAWQHdfsClientConfigurations
+    resultItems = self.validateConfigurationsForSite(stackAdvisor, configurations, recommendedDefaults, services, hosts, siteName, method)
+    items.extend(resultItems)
+    return items
+
+  def isHawqMasterPortConflict(self, configurations):
+    prop_name = 'hawq_master_address_port'
+    default_ambari_port = 5432
+    if prop_name in configurations["hawq-site"]["properties"]:
+      portValue = int(configurations["hawq-site"]["properties"][prop_name])
+      return portValue == default_ambari_port
+
+    return False
+
+  def validateIfRootDir(self, properties, validationItems, prop_name, display_name):
+    root_dir = '/'
+    if prop_name in properties and properties[prop_name].strip() == root_dir:
+      validationItems.append({"config-name": prop_name,
+                              "item": self.getWarnItem(
+                              "It is not advisable to have " + display_name + " at " + root_dir +". Consider creating a sub directory for HAWQ")})
+
+  def checkForMultipleDirs(self, properties, validationItems, prop_name, display_name):
+    # check for delimiters space, comma, colon and semi-colon
+    if prop_name in properties and len(re.sub(r'[,;:]', ' ', properties[prop_name]).split(' ')) > 1:
+      validationItems.append({"config-name": prop_name,
+                              "item": self.getErrorItem(
+                              "Multiple directories for " + display_name + " are not allowed.")})
+
+  def validateHAWQSiteConfigurations(self, stackAdvisor, properties, recommendedDefaults, configurations, services, hosts):
+
+    hawq_site = properties
+    validationItems = []
+
+    # 1. Check if HAWQ master/standby port numbers don't conflict with Ambari ports. Both Ambari and HAWQ use postgres DB and 5432 port.
+    if self.isHawqMasterComponentOnAmbariServer(stackAdvisor, services) and self.isHawqMasterPortConflict(configurations):
+      prop_name = 'hawq_master_address_port'
+      validationItems.append({"config-name": prop_name,
+                                "item": self.getWarnItem(
+                                "The default Postgres port (5432) on the Ambari Server conflicts with the default HAWQ Masters port. "
+                                "If you are using port 5432 for Postgres, you must either deploy the HAWQ Masters on a different host "
+                                "or configure a different port for the HAWQ Masters in the HAWQ Configuration page.")})
+
+    # 2. Check if any data directories are pointing to root dir '/'
+    directories = {
+                    'hawq_master_directory': 'HAWQ Master directory',
+                    'hawq_master_temp_directory': 'HAWQ Master temp directory',
+                    'hawq_segment_directory': 'HAWQ Segment directory',
+                    'hawq_segment_temp_directory': 'HAWQ Segment temp directory'
+                  }
+    for property_name, display_name in directories.iteritems():
+      self.validateIfRootDir(properties, validationItems, property_name, display_name)
+
+    # 2.1 Check if any master or segment directories has multiple values
+    directories = {
+                    'hawq_master_directory': 'HAWQ Master directory',
+                    'hawq_segment_directory': 'HAWQ Segment directory'
+                  }
+    for property_name, display_name in directories.iteritems():
+      self.checkForMultipleDirs(properties, validationItems, property_name, display_name)
+
+    # 3. Check YARN RM address properties
+    YARN = "YARN"
+    servicesList = [service["StackServices"]["service_name"] for service in services["services"]]
+    if YARN in servicesList and "yarn-site" in configurations:
+      yarn_site = self.getSiteProperties(configurations, "yarn-site")
+      for hs_prop, ys_prop in self.getHAWQYARNPropertyMapping().items():
+        if hs_prop in hawq_site and ys_prop in yarn_site and hawq_site[hs_prop] != yarn_site[ys_prop]:
+          message = "Expected value: {0} (this property should have the same value as the property {1} in yarn-site)".format(yarn_site[ys_prop], ys_prop)
+          validationItems.append({"config-name": hs_prop, "item": self.getWarnItem(message)})
+
+    # 4. Check HAWQ Resource Manager type
+    HAWQ_GLOBAL_RM_TYPE = "hawq_global_rm_type"
+    if YARN not in servicesList and HAWQ_GLOBAL_RM_TYPE in hawq_site and hawq_site[HAWQ_GLOBAL_RM_TYPE].upper() == YARN:
+      message = "{0} must be set to none if YARN service is not installed".format(HAWQ_GLOBAL_RM_TYPE)
+      validationItems.append({"config-name": HAWQ_GLOBAL_RM_TYPE, "item": self.getErrorItem(message)})
+
+    # 5. Check query limits
+    if ("default_hash_table_bucket_number" in hawq_site and
+        "hawq_rm_nvseg_perquery_limit"     in hawq_site and
+        int(hawq_site["default_hash_table_bucket_number"]) > int(hawq_site["hawq_rm_nvseg_perquery_limit"])):
+      message = "Default buckets for Hash Distributed tables parameter value should not be greater than the value of Virtual Segments Limit per Query (Total) parameter, currently set to {0}.".format(hawq_site["hawq_rm_nvseg_perquery_limit"])
+      validationItems.append({"config-name": "default_hash_table_bucket_number", "item": self.getErrorItem(message)})
+
+    return stackAdvisor.toConfigurationValidationProblems(validationItems, "hawq-site")
+
+  def validateHAWQHdfsClientConfigurations(self, stackAdvisor, properties, recommendedDefaults, configurations, services, hosts):
+    hdfs_client = properties
+    validationItems = []
+
+    # check HAWQ hdfs-client output.replace-datanode-on-failure property
+    PROP_NAME = "output.replace-datanode-on-failure"
+    if PROP_NAME in hdfs_client:
+      value = hdfs_client[PROP_NAME].upper()
+      componentsListList = [service["components"] for service in services["services"]]
+      componentsList = [item["StackServiceComponents"] for sublist in componentsListList for item in sublist]
+      numSegments = len(self.getHosts(componentsList, "HAWQSEGMENT"))
+
+      message = None
+      limit = 3
+      if numSegments > limit and value != 'TRUE':
+        message = "{0} should be set to true (checked) for clusters with more than {1} HAWQ Segments"
+      elif numSegments <= limit and value != 'FALSE':
+        message = "{0} should be set to false (unchecked) for clusters with {1} or less HAWQ Segments"
+
+      if message:
+        validationItems.append({"config-name": PROP_NAME, "item": self.getWarnItem(message.format(PROP_NAME, str(limit)))})
+
+    return stackAdvisor.toConfigurationValidationProblems(validationItems, "hdfs-client")

ambari-server/src/main/resources/common-services/PXF/3.0.0/service_advisor.py

@@ -0,0 +1,71 @@
+#!/usr/bin/env ambari-python-wrap
+"""
+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.
+"""
+import os
+import fnmatch
+import imp
+import socket
+import sys
+import traceback
+
+SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
+STACKS_DIR = os.path.join(SCRIPT_DIR, '../../../stacks/')
+PARENT_FILE = os.path.join(STACKS_DIR, 'service_advisor.py')
+
+try:
+  with open(PARENT_FILE, 'rb') as fp:
+    service_advisor = imp.load_module('service_advisor', fp, PARENT_FILE, ('.py', 'rb', imp.PY_SOURCE))
+except Exception as e:
+  traceback.print_exc()
+  print "Failed to load parent"
+
+class PXF300ServiceAdvisor(service_advisor.ServiceAdvisor):
+
+  def colocateService(self, stackAdvisor, hostsComponentsMap, serviceComponents):
+    # colocate PXF with NAMENODE and DATANODE, if no hosts have been allocated for PXF
+    pxf = [component for component in serviceComponents if component["StackServiceComponents"]["component_name"] == "PXF"][0]
+    if not stackAdvisor.isComponentHostsPopulated(pxf):
+      for hostName in hostsComponentsMap.keys():
+        hostComponents = hostsComponentsMap[hostName]
+        if ({"name": "NAMENODE"} in hostComponents or {"name": "DATANODE"} in hostComponents) \
+            and {"name": "PXF"} not in hostComponents:
+          hostsComponentsMap[hostName].append({ "name": "PXF" })
+        if ({"name": "NAMENODE"} not in hostComponents and {"name": "DATANODE"} not in hostComponents) \
+            and {"name": "PXF"} in hostComponents:
+          hostsComponentsMap[hostName].remove({"name": "PXF"})
+
+  def getComponentLayoutValidations(self, stackAdvisor, services, hosts):
+    componentsListList = [service["components"] for service in services["services"]]
+    componentsList = [item["StackServiceComponents"] for sublist in componentsListList for item in sublist]
+    hostsList = [host["Hosts"]["host_name"] for host in hosts["items"]]
+    hostsCount = len(hostsList)
+
+    pxfHosts = self.getHosts(componentsList, "PXF")
+    expectedPxfHosts = set(self.getHosts(componentsList, "NAMENODE") + self.getHosts(componentsList, "DATANODE"))
+
+    items = []
+
+    # Generate WARNING if any PXF is not colocated with NAMENODE or DATANODE
+    mismatchHosts = sorted(expectedPxfHosts.symmetric_difference(set(pxfHosts)))
+    if len(mismatchHosts) > 0:
+      hostsString = ', '.join(mismatchHosts)
+      message = "PXF must be installed on the NameNode, Standby NameNode and all DataNodes. " \
+                "The following {0} host(s) do not satisfy the colocation recommendation: {1}".format(len(mismatchHosts), hostsString)
+      items.append( { "type": 'host-component', "level": 'WARN', "message": message, "component-name": 'PXF' } )
+
+    return items

ambari-server/src/main/resources/stacks/HDP/2.0.6/services/stack_advisor.py

@@ -29,7 +29,7 @@ class HDP206StackAdvisor(DefaultStackAdvisor):
 
   def getComponentLayoutValidations(self, services, hosts):
     """Returns array of Validation objects about issues with hostnames components assigned to"""
-    items = []
+    items = super(HDP206StackAdvisor, self).getComponentLayoutValidations(services, hosts)
 
     # Validating NAMENODE and SECONDARY_NAMENODE are on different hosts if possible
     # Use a set for fast lookup
@@ -926,39 +926,6 @@ class HDP206StackAdvisor(DefaultStackAdvisor):
 
     return cluster
 
-  def getConfigurationsValidationItems(self, services, hosts):
-    """Returns array of Validation objects about issues with configuration values provided in services"""
-    items = []
-
-    recommendations = self.recommendConfigurations(services, hosts)
-    recommendedDefaults = recommendations["recommendations"]["blueprint"]["configurations"]
-
-    configurations = services["configurations"]
-    for service in services["services"]:
-      serviceName = service["StackServices"]["service_name"]
-      validator = self.validateServiceConfigurations(serviceName)
-      if validator is not None:
-        for siteName, method in validator.items():
-          if siteName in recommendedDefaults:
-            siteProperties = getSiteProperties(configurations, siteName)
-            if siteProperties is not None:
-              siteRecommendations = recommendedDefaults[siteName]["properties"]
-              print("SiteName: %s, method: %s\n" % (siteName, method.__name__))
-              print("Site properties: %s\n" % str(siteProperties))
-              print("Recommendations: %s\n********\n" % str(siteRecommendations))
-              resultItems = method(siteProperties, siteRecommendations, configurations, services, hosts)
-              items.extend(resultItems)
-
-    clusterWideItems = self.validateClusterConfigurations(configurations, services, hosts)
-    items.extend(clusterWideItems)
-    self.validateMinMax(items, recommendedDefaults, configurations)
-    return items
-
-  def validateClusterConfigurations(self, configurations, services, hosts):
-    validationItems = []
-
-    return self.toConfigurationValidationProblems(validationItems, "")
-
   def getServiceConfigurationValidators(self):
     return {
       "HDFS": { "hdfs-site": self.validateHDFSConfigurations,
@@ -1282,16 +1249,6 @@ class HDP206StackAdvisor(DefaultStackAdvisor):
   def validateServiceConfigurations(self, serviceName):
     return self.getServiceConfigurationValidators().get(serviceName, None)
 
-  def toConfigurationValidationProblems(self, validationProblems, siteName):
-    result = []
-    for validationProblem in validationProblems:
-      validationItem = validationProblem.get("item", None)
-      if validationItem is not None:
-        problem = {"type": 'configuration', "level": validationItem["level"], "message": validationItem["message"],
-                   "config-type": siteName, "config-name": validationProblem["config-name"] }
-        result.append(problem)
-    return result
-
   def getWarnItem(self, message):
     return {"level": "WARN", "message": message}
 

ambari-server/src/main/resources/stacks/HDP/2.3/services/stack_advisor.py

@@ -26,44 +26,6 @@ DB_TYPE_DEFAULT_PORT_MAP = {"MYSQL":"3306", "ORACLE":"1521", "POSTGRES":"5432",
 
 class HDP23StackAdvisor(HDP22StackAdvisor):
 
-  def createComponentLayoutRecommendations(self, services, hosts):
-    parentComponentLayoutRecommendations = super(HDP23StackAdvisor, self).createComponentLayoutRecommendations(services, hosts)
-
-    hostsList = [host["Hosts"]["host_name"] for host in hosts["items"]]
-    hostGroups = parentComponentLayoutRecommendations["blueprint"]["host_groups"]
-    servicesList = [service["StackServices"]["service_name"] for service in services["services"]]
-    componentsList = [component for service in services["services"] for component in service["components"]]
-
-    if "HAWQ" in servicesList:
-      # remove HAWQSTANDBY on a single node
-      if len(hostsList) == 1:
-        components = parentComponentLayoutRecommendations["blueprint"]["host_groups"][0]["components"]
-        components = [component for component in components if component["name"] != 'HAWQSTANDBY']
-        parentComponentLayoutRecommendations["blueprint"]["host_groups"][0]["components"] = components
-
-      # co-locate HAWQSEGMENT with DATANODE, if no hosts have been allocated for HAWQSEGMENT
-      hawqSegment = [component for component in componentsList if component["StackServiceComponents"]["component_name"] == "HAWQSEGMENT"][0]
-      if not self.isComponentHostsPopulated(hawqSegment):
-        for host_group in hostGroups:
-          if {"name": "DATANODE"} in host_group["components"] and {"name": "HAWQSEGMENT"} not in host_group["components"]:
-            host_group["components"].append({"name": "HAWQSEGMENT"})
-          if {"name": "DATANODE"} not in host_group["components"] and {"name": "HAWQSEGMENT"} in host_group["components"]:
-            host_group["components"].remove({"name": "HAWQSEGMENT"})
-
-    if "PXF" in servicesList:
-      # co-locate PXF with NAMENODE and DATANODE, if no hosts have been allocated for PXF
-      pxf = [component for component in componentsList if component["StackServiceComponents"]["component_name"] == "PXF"][0]
-      if not self.isComponentHostsPopulated(pxf):
-        for host_group in hostGroups:
-          if ({"name": "NAMENODE"} in host_group["components"] or {"name": "DATANODE"} in host_group["components"]) \
-              and {"name": "PXF"} not in host_group["components"]:
-            host_group["components"].append({"name": "PXF"})
-          if ({"name": "NAMENODE"} not in host_group["components"] and {"name": "DATANODE"} not in host_group["components"]) \
-              and {"name": "PXF"} in host_group["components"]:
-            host_group["components"].remove({"name": "PXF"})
-
-    return parentComponentLayoutRecommendations
-
   def getComponentLayoutValidations(self, services, hosts):
     parentItems = super(HDP23StackAdvisor, self).getComponentLayoutValidations(services, hosts)
 
@@ -72,52 +34,6 @@ class HDP23StackAdvisor(HDP22StackAdvisor):
     componentsList = [item["StackServiceComponents"] for sublist in componentsListList for item in sublist]
     childItems = []
 
-    if "HAWQ" in servicesList:
-      hostsList = [host["Hosts"]["host_name"] for host in hosts["items"]]
-      hostsCount = len(hostsList)
-
-      hawqMasterHosts = self.__getHosts(componentsList, "HAWQMASTER")
-      hawqStandbyHosts = self.__getHosts(componentsList, "HAWQSTANDBY")
-      hawqSegmentHosts = self.__getHosts(componentsList, "HAWQSEGMENT")
-      datanodeHosts = self.__getHosts(componentsList, "DATANODE")
-
-      # Generate WARNING if any HAWQSEGMENT is not colocated with a DATANODE
-      mismatchHosts = sorted(set(hawqSegmentHosts).symmetric_difference(set(datanodeHosts)))
-      if len(mismatchHosts) > 0:
-        hostsString = ', '.join(mismatchHosts)
-        message = "HAWQ Segment must be installed on all DataNodes. " \
-                  "The following {0} host(s) do not satisfy the colocation recommendation: {1}".format(len(mismatchHosts), hostsString)
-        childItems.append( { "type": 'host-component', "level": 'WARN', "message": message, "component-name": 'HAWQSEGMENT' } )
-
-      # single node case is not analyzed because HAWQ Standby Master will not be present in single node topology due to logic in createComponentLayoutRecommendations()
-      if len(hawqMasterHosts) == 1 and len(hawqStandbyHosts) == 1 and hawqMasterHosts == hawqStandbyHosts:
-        message = "HAWQ Master and HAWQ Standby Master cannot be deployed on the same host."
-        childItems.append( { "type": 'host-component', "level": 'ERROR', "message": message, "component-name": 'HAWQSTANDBY', "host": hawqStandbyHosts[0] } )
-
-      if len(hawqMasterHosts) ==  1 and hostsCount > 1 and self.isLocalHost(hawqMasterHosts[0]):
-        message = "The default Postgres port (5432) on the Ambari Server conflicts with the default HAWQ Masters port. " \
-                  "If you are using port 5432 for Postgres, you must either deploy the HAWQ Master on a different host " \
-                  "or configure a different port for the HAWQ Masters in the HAWQ Configuration page."
-        childItems.append( { "type": 'host-component', "level": 'WARN', "message": message, "component-name": 'HAWQMASTER', "host": hawqMasterHosts[0] } )
-
-      if len(hawqStandbyHosts) ==  1 and hostsCount > 1 and self.isLocalHost(hawqStandbyHosts[0]):
-        message = "The default Postgres port (5432) on the Ambari Server conflicts with the default HAWQ Masters port. " \
-                  "If you are using port 5432 for Postgres, you must either deploy the HAWQ Standby Master on a different host " \
-                  "or configure a different port for the HAWQ Masters in the HAWQ Configuration page."
-        childItems.append( { "type": 'host-component', "level": 'WARN', "message": message, "component-name": 'HAWQSTANDBY', "host": hawqStandbyHosts[0] } )
-
-    if "PXF" in servicesList:
-      pxfHosts = self.__getHosts(componentsList, "PXF")
-      expectedPxfHosts = set(self.__getHosts(componentsList, "NAMENODE") + self.__getHosts(componentsList, "DATANODE"))
-
-      # Generate WARNING if any PXF is not colocated with NAMENODE or DATANODE
-      mismatchHosts = sorted(expectedPxfHosts.symmetric_difference(set(pxfHosts)))
-      if len(mismatchHosts) > 0:
-        hostsString = ', '.join(mismatchHosts)
-        message = "PXF must be installed on the NameNode, Standby NameNode and all DataNodes. " \
-                  "The following {0} host(s) do not satisfy the colocation recommendation: {1}".format(len(mismatchHosts), hostsString)
-        childItems.append( { "type": 'host-component', "level": 'WARN', "message": message, "component-name": 'PXF' } )
-
     if "SPARK" in servicesList:
       if "SPARK_THRIFTSERVER" in servicesList:
         if not "HIVE_SERVER" in servicesList:
@@ -143,21 +59,6 @@ class HDP23StackAdvisor(HDP22StackAdvisor):
     else:
       return []
 
-  def getNotPreferableOnServerComponents(self):
-    parentComponents = super(HDP23StackAdvisor, self).getNotPreferableOnServerComponents()
-    parentComponents.extend(['HAWQMASTER', 'HAWQSTANDBY'])
-    return parentComponents
-
-  def getComponentLayoutSchemes(self):
-    parentSchemes = super(HDP23StackAdvisor, self).getComponentLayoutSchemes()
-    # key is max number of cluster hosts + 1, value is index in host list where to put the component
-    childSchemes = {
-        'HAWQMASTER' : {6: 2, 31: 1, "else": 5},
-        'HAWQSTANDBY': {6: 1, 31: 2, "else": 3}
-    }
-    parentSchemes.update(childSchemes)
-    return parentSchemes
-
   def getServiceConfigurationRecommenderDict(self):
     parentRecommendConfDict = super(HDP23StackAdvisor, self).getServiceConfigurationRecommenderDict()
     childRecommendConfDict = {
@@ -169,7 +70,6 @@ class HDP23StackAdvisor(HDP22StackAdvisor):
       "KAFKA": self.recommendKAFKAConfigurations,
       "RANGER": self.recommendRangerConfigurations,
       "RANGER_KMS": self.recommendRangerKMSConfigurations,
-      "HAWQ": self.recommendHAWQConfigurations,
       "FALCON": self.recommendFalconConfigurations,
       "STORM": self.recommendStormConfigurations,
       "SQOOP": self.recommendSqoopConfigurations,
@@ -715,59 +615,6 @@ class HDP23StackAdvisor(HDP22StackAdvisor):
         putYarnSiteProperty('yarn.resourcemanager.proxy-user-privileges.enabled', 'false')
 
 
-  def isHawqMasterComponentOnAmbariServer(self, services):
-    componentsListList = [service["components"] for service in services["services"]]
-    componentsList = [item for sublist in componentsListList for item in sublist]
-    hawqMasterComponentHosts = [hostname for component in componentsList if component["StackServiceComponents"]["component_name"] in ("HAWQMASTER", "HAWQSTANDBY") for hostname in component["StackServiceComponents"]["hostnames"]]
-    return any([self.isLocalHost(host) for host in hawqMasterComponentHosts])
-
-
-  def recommendHAWQConfigurations(self, configurations, clusterData, services, hosts):
-    if any(x in services["configurations"] for x in ["hawq-site", "hdfs-client"]):
-      componentsListList = [service["components"] for service in services["services"]]
-      componentsList = [item["StackServiceComponents"] for sublist in componentsListList for item in sublist]
-      servicesList = [service["StackServices"]["service_name"] for service in services["services"]]
-      numSegments = len(self.__getHosts(componentsList, "HAWQSEGMENT"))
-
-    if "hawq-site" in services["configurations"]:
-      hawq_site = services["configurations"]["hawq-site"]["properties"]
-      putHawqSiteProperty = self.putProperty(configurations, "hawq-site", services)
-
-      # remove master port when master is colocated with Ambari server
-      if self.isHawqMasterComponentOnAmbariServer(services) and "hawq_master_address_port" in hawq_site:
-        putHawqSiteProperty('hawq_master_address_port', '')
-
-      # update query limits if segments are deployed
-      if numSegments and "default_hash_table_bucket_number" in hawq_site and "hawq_rm_nvseg_perquery_limit" in hawq_site:
-        factor_min = 1
-        factor_max = 6
-        limit = int(hawq_site["hawq_rm_nvseg_perquery_limit"])
-        factor = limit / numSegments
-        # if too many segments or default limit is too low --> stick with the limit
-        if factor < factor_min:
-          buckets = limit
-        # if the limit is large and results in factor > max --> limit factor to max
-        elif factor > factor_max:
-          buckets = factor_max * numSegments
-        else:
-          buckets = factor * numSegments
-        putHawqSiteProperty('default_hash_table_bucket_number', buckets)
-
-      # update YARN RM urls with the values from yarn-site if YARN is installed
-      if "YARN" in servicesList and "yarn-site" in services["configurations"]:
-        yarn_site = services["configurations"]["yarn-site"]["properties"]
-        for hs_prop, ys_prop in self.getHAWQYARNPropertyMapping().items():
-          if hs_prop in hawq_site and ys_prop in yarn_site:
-            putHawqSiteProperty(hs_prop, yarn_site[ys_prop])
-
-    # set output.replace-datanode-on-failure in HAWQ hdfs-client depending on the cluster size
-    if "hdfs-client" in services["configurations"]:
-      hdfs_client = services["configurations"]["hdfs-client"]["properties"]
-      if "output.replace-datanode-on-failure" in hdfs_client:
-        propertyValue = "true" if numSegments > 3 else "false"
-        putHdfsClientProperty = self.putProperty(configurations, "hdfs-client", services)
-        putHdfsClientProperty("output.replace-datanode-on-failure", propertyValue)
-
   def recommendSqoopConfigurations(self, configurations, clusterData, services, hosts):
     putSqoopSiteProperty = self.putProperty(configurations, "sqoop-site", services)
 
@@ -861,9 +708,7 @@ class HDP23StackAdvisor(HDP22StackAdvisor):
                "hive-site": self.validateHiveConfigurations},
       "HBASE": {"hbase-site": self.validateHBASEConfigurations},
       "KAKFA": {"kafka-broker": self.validateKAFKAConfigurations},
-      "YARN": {"yarn-site": self.validateYARNConfigurations},
-      "HAWQ": {"hawq-site": self.validateHAWQSiteConfigurations,
-               "hdfs-client": self.validateHAWQHdfsClientConfigurations}
+      "YARN": {"yarn-site": self.validateYARNConfigurations}
     }
     self.mergeValidators(parentValidators, childValidators)
     return parentValidators
@@ -1062,115 +907,5 @@ class HDP23StackAdvisor(HDP22StackAdvisor):
 
     return self.toConfigurationValidationProblems(validationItems, "yarn-site")
 
-
-  def isHawqMasterPortConflict(self, configurations):
-    prop_name = 'hawq_master_address_port'
-    default_ambari_port = 5432
-    if prop_name in configurations["hawq-site"]["properties"]:
-      portValue = int(configurations["hawq-site"]["properties"][prop_name])
-      return portValue == default_ambari_port
-
-    return False
-
-
-  def validateIfRootDir(self, properties, validationItems, prop_name, display_name):
-    root_dir = '/'
-    if prop_name in properties and properties[prop_name].strip() == root_dir:
-      validationItems.append({"config-name": prop_name,
-                              "item": self.getWarnItem(
-                              "It is not advisable to have " + display_name + " at " + root_dir +". Consider creating a sub directory for HAWQ")})
-
-
-  def checkForMultipleDirs(self, properties, validationItems, prop_name, display_name):
-    # check for delimiters space, comma, colon and semi-colon
-    if prop_name in properties and len(re.sub(r'[,;:]', ' ', properties[prop_name]).split(' ')) > 1:
-      validationItems.append({"config-name": prop_name,
-                              "item": self.getErrorItem(
-                              "Multiple directories for " + display_name + " are not allowed.")})
-
-
-  def validateHAWQSiteConfigurations(self, properties, recommendedDefaults, configurations, services, hosts):
-    hawq_site = properties
-    validationItems = []
-
-    # 1. Check if HAWQ master/standby port numbers don't conflict with Ambari ports. Both Ambari and HAWQ use postgres DB and 5432 port.
-    if self.isHawqMasterComponentOnAmbariServer(services) and self.isHawqMasterPortConflict(configurations):
-      prop_name = 'hawq_master_address_port'
-      validationItems.append({"config-name": prop_name,
-                                "item": self.getWarnItem(
-                                "The default Postgres port (5432) on the Ambari Server conflicts with the default HAWQ Masters port. "
-                                "If you are using port 5432 for Postgres, you must either deploy the HAWQ Masters on a different host "
-                                "or configure a different port for the HAWQ Masters in the HAWQ Configuration page.")})
-
-    # 2. Check if any data directories are pointing to root dir '/'
-    directories = {
-                    'hawq_master_directory': 'HAWQ Master directory',
-                    'hawq_master_temp_directory': 'HAWQ Master temp directory',
-                    'hawq_segment_directory': 'HAWQ Segment directory',
-                    'hawq_segment_temp_directory': 'HAWQ Segment temp directory'
-                  }
-    for property_name, display_name in directories.iteritems():
-      self.validateIfRootDir(properties, validationItems, property_name, display_name)
-
-    # 2.1 Check if any master or segment directories has multiple values
-    directories = {
-                    'hawq_master_directory': 'HAWQ Master directory',
-                    'hawq_segment_directory': 'HAWQ Segment directory'
-                  }
-    for property_name, display_name in directories.iteritems():
-      self.checkForMultipleDirs(properties, validationItems, property_name, display_name)
-
-    # 3. Check YARN RM address properties
-    YARN = "YARN"
-    servicesList = [service["StackServices"]["service_name"] for service in services["services"]]
-    if YARN in servicesList and "yarn-site" in configurations:
-      yarn_site = getSiteProperties(configurations, "yarn-site")
-      for hs_prop, ys_prop in self.getHAWQYARNPropertyMapping().items():
-        if hs_prop in hawq_site and ys_prop in yarn_site and hawq_site[hs_prop] != yarn_site[ys_prop]:
-          message = "Expected value: {0} (this property should have the same value as the property {1} in yarn-site)".format(yarn_site[ys_prop], ys_prop)
-          validationItems.append({"config-name": hs_prop, "item": self.getWarnItem(message)})
-
-    # 4. Check HAWQ Resource Manager type
-    HAWQ_GLOBAL_RM_TYPE = "hawq_global_rm_type"
-    if YARN not in servicesList and HAWQ_GLOBAL_RM_TYPE in hawq_site and hawq_site[HAWQ_GLOBAL_RM_TYPE].upper() == YARN:
-      message = "{0} must be set to none if YARN service is not installed".format(HAWQ_GLOBAL_RM_TYPE)
-      validationItems.append({"config-name": HAWQ_GLOBAL_RM_TYPE, "item": self.getErrorItem(message)})
-
-    # 5. Check query limits
-    if ("default_hash_table_bucket_number" in hawq_site and
-        "hawq_rm_nvseg_perquery_limit"     in hawq_site and
-        int(hawq_site["default_hash_table_bucket_number"]) > int(hawq_site["hawq_rm_nvseg_perquery_limit"])):
-      message = "Default buckets for Hash Distributed tables parameter value should not be greater than the value of Virtual Segments Limit per Query (Total) parameter, currently set to {0}.".format(hawq_site["hawq_rm_nvseg_perquery_limit"])
-      validationItems.append({"config-name": "default_hash_table_bucket_number", "item": self.getErrorItem(message)})
-
-    return self.toConfigurationValidationProblems(validationItems, "hawq-site")
-
-  def validateHAWQHdfsClientConfigurations(self, properties, recommendedDefaults, configurations, services, hosts):
-    hdfs_client = properties
-    validationItems = []
-
-    # check HAWQ hdfs-client output.replace-datanode-on-failure property
-    PROP_NAME = "output.replace-datanode-on-failure"
-    if PROP_NAME in hdfs_client:
-      value = hdfs_client[PROP_NAME].upper()
-      componentsListList = [service["components"] for service in services["services"]]
-      componentsList = [item["StackServiceComponents"] for sublist in componentsListList for item in sublist]
-      numSegments = len(self.__getHosts(componentsList, "HAWQSEGMENT"))
-
-      message = None
-      limit = 3
-      if numSegments > limit and value != 'TRUE':
-        message = "{0} should be set to true (checked) for clusters with more than {1} HAWQ Segments"
-      elif numSegments <= limit and value != 'FALSE':
-        message = "{0} should be set to false (unchecked) for clusters with {1} or less HAWQ Segments"
-
-      if message:
-        validationItems.append({"config-name": PROP_NAME, "item": self.getWarnItem(message.format(PROP_NAME, str(limit)))})
-
-    return self.toConfigurationValidationProblems(validationItems, "hdfs-client")
-
   def isComponentUsingCardinalityForLayout(self, componentName):
     return componentName in ['NFS_GATEWAY', 'PHOENIX_QUERY_SERVER', 'SPARK_THRIFTSERVER']
-
-  def getHAWQYARNPropertyMapping(self):
-    return { "hawq_rm_yarn_address": "yarn.resourcemanager.address", "hawq_rm_yarn_scheduler_address": "yarn.resourcemanager.scheduler.address" }

ambari-server/src/main/resources/stacks/service_advisor.py

@@ -0,0 +1,247 @@
+#!/usr/bin/env ambari-python-wrap
+"""
+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.
+"""
+
+"""
+The naming convention for ServiceAdvisor subclasses depends on whether they are
+in common-services or are part of the stack version's services.
+
+In common-services, the naming convention is <service_name><service_version>ServiceAdvisor.
+In the stack, the naming convention is <stack_name><stack_version><service_name>ServiceAdvisor.
+
+Unlike the StackAdvisor, the ServiceAdvisor does NOT provide any inheritance.
+If you want to use inheritance to augment a previous version of a service's
+advisor you can use the following code to dynamically load the previous advisor.
+Some changes will be need to provide the correct path and class names.
+
+  SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
+  PARENT_DIR = os.path.join(SCRIPT_DIR, '../<old_version>')
+  PARENT_FILE = os.path.join(PARENT_DIR, 'service_advisor.py')
+
+  try:
+    with open(PARENT_FILE, 'rb') as fp:
+      service_advisor = imp.load_module('service_advisor', fp, PARENT_FILE, ('.py', 'rb', imp.PY_SOURCE))
+  except Exception as e:
+    traceback.print_exc()
+    print "Failed to load parent"
+
+  class <NewServiceAdvisorClassName>(service_advisor.<OldServiceAdvisorClassName>)
+
+where the NewServiceAdvisorClassName and OldServiceAdvisorClassName follow the naming
+convention listed above.
+
+For examples see: common-services/HAWQ/2.0.0/service_advisor.py
+and common-services/PXF/3.0.0/service_advisor.py
+"""
+class ServiceAdvisor(object):
+  """
+  Abstract class implemented by all service advisors.
+  """
+
+  """
+  Provides a scheme for laying out a given component on different number of hosts.
+  It should return a json object such as {6: 2, 31: 1, "else": 5}.  If a service
+  contains more than one component, it may have a statement such as:
+
+    if componentName == 'HAWQMASTER':
+      return {6: 2, 31: 1, "else": 5}
+
+    return None
+  """
+  def getComponentLayoutScheme(self, componentName):
+    return None
+
+  """
+  Returns the cardinality of the component as a json object with min and max attributes.
+  Example: {"min": 1, "max": 1}
+  """
+  def getComponentCardinality(self, componentName):
+    return {"min": 1, "max": 1}
+
+  """
+  Returns True if the component should avoid being configured on the ambari server host.
+  """
+  def isComponentNotPreferableOnAmbariServerHost(self, componentName):
+    return False
+
+  """
+  Return True if the component is not considered valuable, otherwise returns False.
+  Hosts which are only running non valuable components are considered to be not used.
+  """
+  def isComponentNotValuable(self, component):
+    return False
+
+  """
+  Returns True if the component should use its cardinatity to determine its layout.
+  """
+  def isComponentUsingCardinalityForLayout(self, componentName):
+    return False
+
+  """
+  Returns True if the component is a master component with multiple instances.
+  """
+  def isMasterComponentWithMultipleInstances(self, component):
+    return False
+
+  """
+  If any components of the service should be colocated with other services,
+  this is where you should set up that layout.  Example:
+
+    # colocate HAWQSEGMENT with DATANODE, if no hosts have been allocated for HAWQSEGMENT
+    hawqSegment = [component for component in serviceComponents if component["StackServiceComponents"]["component_name"] == "HAWQSEGMENT"][0]
+    if not stackAdvisor.isComponentHostsPopulated(hawqSegment):
+      for hostName in hostsComponentsMap.keys():
+        hostComponents = hostsComponentsMap[hostName]
+        if {"name": "DATANODE"} in hostComponents and {"name": "HAWQSEGMENT"} not in hostComponents:
+          hostsComponentsMap[hostName].append( { "name": "HAWQSEGMENT" } )
+        if {"name": "DATANODE"} not in hostComponents and {"name": "HAWQSEGMENT"} in hostComponents:
+          hostComponents.remove({"name": "HAWQSEGMENT"})
+  """
+  def colocateService(self, stackAdvisor, hostsComponentsMap, serviceComponents):
+    pass
+
+  """
+  Any configuration recommendations for the service should be defined in this function.
+  This should be similar to any of the recommendXXXXConfigurations functions in the stack_advisor.py
+  such as recommendYARNConfigurations().
+  """
+  def getServiceConfigurationRecommendations(self, stackAdvisor, configurations, clusterSummary, services, hosts):
+    pass
+
+  """
+  Returns an array of Validation objects about issues with the hostnames to which components are assigned.
+  This should detect validation issues which are different than those the stack_advisor.py detects.
+  The default validations are in stack_advisor.py getComponentLayoutValidations function.
+  """
+  def getComponentLayoutValidations(self, stackAdvisor, services, hosts):
+    return []
+
+  """
+  Any configuration validations for the service should be defined in this function.
+  This should be similar to any of the validateXXXXConfigurations functions in the stack_advisor.py
+  such as validateHDFSConfigurations.
+  """
+  def getConfigurationsValidationItems(self, stackAdvisor, configurations, recommendedDefaults, services, hosts):
+    return []
+
+  """
+  If the service needs to any processing differently than the stack_advisor.py getHostsForMasterComponents
+  function, then this logic should be added in this function.
+  """
+  def getHostsForMasterComponent(self, stackAdvisor, services, hosts, component, hostsList, hostsComponentsMap):
+    return stackAdvisor.getHostsForMasterComponent(services, hosts, component, hostsList, hostsComponentsMap)
+
+  """
+  If the service needs to any processing differently than the stack_advisor.py getHostsForSlaveComponents
+  function, then this logic should be added in this function.
+  """
+  def getHostsForSlaveComponent(self, stackAdvisor, services, hosts, component, hostsList, hostsComponentsMap, freeHosts):
+    return stackAdvisor.getHostsForSlaveComponent(services, hosts, component, hostsList, hostsComponentsMap, freeHosts)
+
+
+  """
+  Utility methods
+  """
+
+  """
+  Utility method used for validation warnings.
+  """
+  def getWarnItem(self, message):
+    return {"level": "WARN", "message": message}
+
+  """
+  Utility method used for validation errors.
+  """
+  def getErrorItem(self, message):
+    return {"level": "ERROR", "message": message}
+
+  """
+  Returns the hosts which are running the given component.
+  """
+  def getHosts(self, componentsList, componentName):
+    return [component["hostnames"] for component in componentsList if component["component_name"] == componentName][0]
+
+  """
+  Utility method for setting a configuration property value.
+  """
+  def putProperty(self, config, configType, services=None):
+    userConfigs = {}
+    changedConfigs = []
+    # if services parameter, prefer values, set by user
+    if services:
+      if 'configurations' in services.keys():
+        userConfigs = services['configurations']
+      if 'changed-configurations' in services.keys():
+        changedConfigs = services["changed-configurations"]
+
+    if configType not in config:
+      config[configType] = {}
+    if"properties" not in config[configType]:
+      config[configType]["properties"] = {}
+    def appendProperty(key, value):
+      # If property exists in changedConfigs, do not override, use user defined property
+      if self.__isPropertyInChangedConfigs(configType, key, changedConfigs):
+        config[configType]["properties"][key] = userConfigs[configType]['properties'][key]
+      else:
+        config[configType]["properties"][key] = str(value)
+    return appendProperty
+
+  """
+  Utility method to determine if the configuration property value has been changed.
+  """
+  def __isPropertyInChangedConfigs(self, configType, propertyName, changedConfigs):
+    for changedConfig in changedConfigs:
+      if changedConfig['type']==configType and changedConfig['name']==propertyName:
+        return True
+    return False
+
+  """
+  Utility method for setting a configuration property attribute.
+  """
+  def putPropertyAttribute(self, config, configType):
+    if configType not in config:
+      config[configType] = {}
+    def appendPropertyAttribute(key, attribute, attributeValue):
+      if "property_attributes" not in config[configType]:
+        config[configType]["property_attributes"] = {}
+      if key not in config[configType]["property_attributes"]:
+        config[configType]["property_attributes"][key] = {}
+      config[configType]["property_attributes"][key][attribute] = attributeValue if isinstance(attributeValue, list) else str(attributeValue)
+    return appendPropertyAttribute
+
+  """
+  Utility method to validate configuration settings for a given configuration file.
+  This function will call the method provided.
+  """
+  def validateConfigurationsForSite(self, stackAdvisor, configurations, recommendedDefaults, services, hosts, siteName, method):
+    if siteName in recommendedDefaults:
+      siteProperties = self.getSiteProperties(configurations, siteName)
+      if siteProperties is not None:
+        siteRecommendations = recommendedDefaults[siteName]["properties"]
+        print("SiteName: %s, method: %s\n" % (siteName, method.__name__))
+        print("Site properties: %s\n" % str(siteProperties))
+        print("Recommendations: %s\n********\n" % str(siteRecommendations))
+        return method(stackAdvisor, siteProperties, siteRecommendations, configurations, services, hosts)
+    return []
+
+  """
+  Utility method used to return all the properties from a given configuration file.
+  """
+  def getSiteProperties(self, configurations, siteName):
+    siteConfig = configurations.get(siteName)
+    if siteConfig is None:
+      return None
+    return siteConfig.get("properties")

ambari-server/src/main/resources/stacks/stack_advisor.py

@@ -309,6 +309,8 @@ class DefaultStackAdvisor(StackAdvisor):
   implement
   """
 
+  services = None
+
   """
   Filters the list of specified hosts object and returns
   a list of hosts which are not in maintenance mode.
@@ -342,6 +344,7 @@ class DefaultStackAdvisor(StackAdvisor):
     return recommendations
 
   def createComponentLayoutRecommendations(self, services, hosts):
+    self.services = services
 
     recommendations = {
       "blueprint": {
@@ -365,27 +368,14 @@ class DefaultStackAdvisor(StackAdvisor):
     #extend hostsComponentsMap' with MASTER components
     for service in services["services"]:
       masterComponents = [component for component in service["components"] if self.isMasterComponent(component)]
+      serviceAdvisor = self.instantiateServiceAdvisor(service)
       for component in masterComponents:
         componentName = component["StackServiceComponents"]["component_name"]
-
-        if self.isComponentHostsPopulated(component):
-          hostsForComponent = component["StackServiceComponents"]["hostnames"]
+        hostsForComponent = []
+        if serviceAdvisor is None:
+          hostsForComponent = self.getHostsForMasterComponent(services, hosts, component, hostsList, hostsComponentsMap)
         else:
-
-          if len(hostsList) > 1 and self.isMasterComponentWithMultipleInstances(component):
-            hostsCount = self.getMinComponentCount(component)
-            if hostsCount > 1: # get first 'hostsCount' available hosts
-              hostsForComponent = []
-              hostIndex = 0
-              while hostsCount > len(hostsForComponent) and hostIndex < len(hostsList):
-                currentHost = hostsList[hostIndex]
-                if self.isHostSuitableForComponent(currentHost, component):
-                  hostsForComponent.append(currentHost)
-                hostIndex += 1
-            else:
-              hostsForComponent = [self.getHostForComponent(component, hostsList)]
-          else:
-            hostsForComponent = [self.getHostForComponent(component, hostsList)]
+          hostsForComponent = serviceAdvisor.getHostsForMasterComponent(self, services, hosts, component, hostsList, hostsComponentsMap)
 
         #extend 'hostsComponentsMap' with 'hostsForComponent'
         for hostName in hostsForComponent:
@@ -402,41 +392,14 @@ class DefaultStackAdvisor(StackAdvisor):
     for service in services["services"]:
       slaveClientComponents = [component for component in service["components"]
                                if self.isSlaveComponent(component) or self.isClientComponent(component)]
+      serviceAdvisor = self.instantiateServiceAdvisor(service)
       for component in slaveClientComponents:
         componentName = component["StackServiceComponents"]["component_name"]
-
-        if component["StackServiceComponents"]["cardinality"] == "ALL":
-          hostsForComponent = hostsList
+        hostsForComponent = []
+        if serviceAdvisor is None:
+          hostsForComponent = self.getHostsForSlaveComponent(services, hosts, component, hostsList, hostsComponentsMap, freeHosts)
         else:
-          componentIsPopulated = self.isComponentHostsPopulated(component)
-          if componentIsPopulated:
-            hostsForComponent = component["StackServiceComponents"]["hostnames"]
-          else:
-            hostsForComponent = []
-
-          if self.isSlaveComponent(component):
-            cardinality = str(component["StackServiceComponents"]["cardinality"])
-            if self.isComponentUsingCardinalityForLayout(componentName) and cardinality:
-              # cardinality types: 1+, 1-2, 1, ALL
-              if "+" in cardinality:
-                hostsMin = int(cardinality[:-1])
-              elif "-" in cardinality:
-                nums = cardinality.split("-")
-                hostsMin = int(nums[0])
-              else:
-                hostsMin = int(cardinality)
-              if hostsMin > len(hostsForComponent):
-                hostsForComponent.extend(freeHosts[0:hostsMin-len(hostsForComponent)])
-            # Components which are already installed, keep the recommendation as the existing layout
-            elif not componentIsPopulated:
-              hostsForComponent.extend(freeHosts)
-              if not hostsForComponent:  # hostsForComponent is empty
-                hostsForComponent = hostsList[-1:]
-            hostsForComponent = list(set(hostsForComponent))  # removing duplicates
-          elif self.isClientComponent(component) and not componentIsPopulated:
-            hostsForComponent = freeHosts[0:1]
-            if not hostsForComponent:  # hostsForComponent is empty
-              hostsForComponent = hostsList[-1:]
+          hostsForComponent = serviceAdvisor.getHostsForSlaveComponent(self, services, hosts, component, hostsList, hostsComponentsMap, freeHosts)
 
         #extend 'hostsComponentsMap' with 'hostsForComponent'
         for hostName in hostsForComponent:
@@ -445,6 +408,13 @@ class DefaultStackAdvisor(StackAdvisor):
           if hostName in hostsSet:
             hostsComponentsMap[hostName].append( { "name": componentName } )
 
+    #colocate custom services
+    for service in services["services"]:
+      serviceAdvisor = self.instantiateServiceAdvisor(service)
+      if serviceAdvisor is not None:
+        serviceComponents = [component for component in service["components"]]
+        serviceAdvisor.colocateService(self, hostsComponentsMap, serviceComponents)
+
     #prepare 'host-group's from 'hostsComponentsMap'
     host_groups = recommendations["blueprint"]["host_groups"]
     bindings = recommendations["blueprint_cluster_binding"]["host_groups"]
@@ -458,7 +428,110 @@ class DefaultStackAdvisor(StackAdvisor):
     return recommendations
   pass
 
+  def getHostsForMasterComponent(self, services, hosts, component, hostsList, hostsComponentsMap):
+    componentName = component["StackServiceComponents"]["component_name"]
+
+    if self.isComponentHostsPopulated(component):
+      return component["StackServiceComponents"]["hostnames"]
+
+    if len(hostsList) > 1 and self.isMasterComponentWithMultipleInstances(component):
+      hostsCount = self.getMinComponentCount(component)
+      if hostsCount > 1: # get first 'hostsCount' available hosts
+        hostsForComponent = []
+        hostIndex = 0
+        while hostsCount > len(hostsForComponent) and hostIndex < len(hostsList):
+          currentHost = hostsList[hostIndex]
+          if self.isHostSuitableForComponent(currentHost, component):
+            hostsForComponent.append(currentHost)
+          hostIndex += 1
+        return hostsForComponent
+
+    return [self.getHostForComponent(component, hostsList)]
+
+  def getHostsForSlaveComponent(self, services, hosts, component, hostsList, hostsComponentsMap, freeHosts):
+    componentName = component["StackServiceComponents"]["component_name"]
+
+    if component["StackServiceComponents"]["cardinality"] == "ALL":
+      return hostsList
+
+    componentIsPopulated = self.isComponentHostsPopulated(component)
+    if componentIsPopulated:
+      return component["StackServiceComponents"]["hostnames"]
+
+    hostsForComponent = []
+
+    if self.isSlaveComponent(component):
+      cardinality = str(component["StackServiceComponents"]["cardinality"])
+      if self.isComponentUsingCardinalityForLayout(component) and cardinality:
+        # cardinality types: 1+, 1-2, 1
+        if "+" in cardinality:
+          hostsMin = int(cardinality[:-1])
+        elif "-" in cardinality:
+          nums = cardinality.split("-")
+          hostsMin = int(nums[0])
+        else:
+          hostsMin = int(cardinality)
+        if hostsMin > len(hostsForComponent):
+          hostsForComponent.extend(freeHosts[0:hostsMin-len(hostsForComponent)])
+
+      else:
+        hostsForComponent.extend(freeHosts)
+        if not hostsForComponent:  # hostsForComponent is empty
+          hostsForComponent = hostsList[-1:]
+      hostsForComponent = list(set(hostsForComponent))  # removing duplicates
+    elif self.isClientComponent(component):
+      hostsForComponent = freeHosts[0:1]
+      if not hostsForComponent:  # hostsForComponent is empty
+        hostsForComponent = hostsList[-1:]
+
+    return hostsForComponent
+
+  def instantiateServiceAdvisorForComponent(self, componentName):
+    if self.services is None:
+      return None
+
+    for service in self.services["services"]:
+      for component in service["components"]:
+        if componentName == self.getComponentName(component):
+          return self.instantiateServiceAdvisor(service)
+
+    return None
+
+  def instantiateServiceAdvisor(self, service):
+    import imp
+    import os
+    import traceback
+
+    serviceName = service["StackServices"]["service_name"]
+    className = service["StackServices"]["advisor_name"] if "advisor_name" in service["StackServices"] else None
+    path = service["StackServices"]["advisor_path"] if "advisor_path" in service["StackServices"] else None
+
+    if path is None or className is None:
+      return None
+
+    if not os.path.exists(path):
+      return None
+
+    try:
+      serviceAdvisor = None
+      with open(path, 'rb') as fp:
+        serviceAdvisor = imp.load_module('service_advisor_impl', fp, path, ('.py', 'rb', imp.PY_SOURCE))
+      if hasattr(serviceAdvisor, className):
+        print "ServiceAdvisor implementation for service {0} was loaded".format(serviceName)
+        clazz = getattr(serviceAdvisor, className)
+        return clazz()
+
+    except Exception as e:
+      traceback.print_exc()
+      print "Failed to load or create ServiceAdvisor implementation for service {0}".format(serviceName)
+
+    return None
+
   def isComponentUsingCardinalityForLayout(self, componentName):
+    serviceAdvisor = self.instantiateServiceAdvisorForComponent(componentName)
+    if serviceAdvisor is not None:
+      return serviceAdvisor.isComponentUsingCardinalityForLayout(componentName)
+
     return False
 
   def createValidationResponse(self, services, validationItems):
@@ -480,17 +553,116 @@ class DefaultStackAdvisor(StackAdvisor):
 
   def validateConfigurations(self, services, hosts):
     """Returns array of Validation objects about issues with hostnames components assigned to"""
+    self.services = services
+
     validationItems = self.getConfigurationsValidationItems(services, hosts)
     return self.createValidationResponse(services, validationItems)
 
   def getComponentLayoutValidations(self, services, hosts):
-    return []
+    self.services = services
+
+    items = []
+    if services is None:
+      return items
+
+    for service in services["services"]:
+      advisor = self.instantiateServiceAdvisor(service)
+      if advisor is not None:
+        items.extend(advisor.getComponentLayoutValidations(self, services, hosts))
+
+    return items
 
   def getConfigurationClusterSummary(self, servicesList, hosts, components, services):
     pass
 
+  def validateClusterConfigurations(self, configurations, services, hosts):
+    validationItems = []
+
+    return self.toConfigurationValidationProblems(validationItems, "")
+
+  def toConfigurationValidationProblems(self, validationProblems, siteName):
+    result = []
+    for validationProblem in validationProblems:
+      validationItem = validationProblem.get("item", None)
+      if validationItem is not None:
+        problem = {"type": 'configuration', "level": validationItem["level"], "message": validationItem["message"],
+                   "config-type": siteName, "config-name": validationProblem["config-name"] }
+        result.append(problem)
+    return result
+
+  def validateServiceConfigurations(self, serviceName):
+    return self.getServiceConfigurationValidators().get(serviceName, None)
+
+  def getServiceConfigurationValidators(self):
+    return {}
+
+  def validateMinMax(self, items, recommendedDefaults, configurations):
+
+    # required for casting to the proper numeric type before comparison
+    def convertToNumber(number):
+      try:
+        return int(number)
+      except ValueError:
+        return float(number)
+
+    for configName in configurations:
+      validationItems = []
+      if configName in recommendedDefaults and "property_attributes" in recommendedDefaults[configName]:
+        for propertyName in recommendedDefaults[configName]["property_attributes"]:
+          if propertyName in configurations[configName]["properties"]:
+            if "maximum" in recommendedDefaults[configName]["property_attributes"][propertyName] and \
+                propertyName in recommendedDefaults[configName]["properties"]:
+              userValue = convertToNumber(configurations[configName]["properties"][propertyName])
+              maxValue = convertToNumber(recommendedDefaults[configName]["property_attributes"][propertyName]["maximum"])
+              if userValue > maxValue:
+                validationItems.extend([{"config-name": propertyName, "item": self.getWarnItem("Value is greater than the recommended maximum of {0} ".format(maxValue))}])
+            if "minimum" in recommendedDefaults[configName]["property_attributes"][propertyName] and \
+                    propertyName in recommendedDefaults[configName]["properties"]:
+              userValue = convertToNumber(configurations[configName]["properties"][propertyName])
+              minValue = convertToNumber(recommendedDefaults[configName]["property_attributes"][propertyName]["minimum"])
+              if userValue < minValue:
+                validationItems.extend([{"config-name": propertyName, "item": self.getWarnItem("Value is less than the recommended minimum of {0} ".format(minValue))}])
+      items.extend(self.toConfigurationValidationProblems(validationItems, configName))
+    pass
+
+
   def getConfigurationsValidationItems(self, services, hosts):
-    return []
+    """Returns array of Validation objects about issues with configuration values provided in services"""
+    items = []
+
+    recommendations = self.recommendConfigurations(services, hosts)
+    recommendedDefaults = recommendations["recommendations"]["blueprint"]["configurations"]
+    configurations = services["configurations"]
+
+    for service in services["services"]:
+      items.extend(self.getConfigurationsValidationItemsForService(configurations, recommendedDefaults, service, services, hosts))
+
+    clusterWideItems = self.validateClusterConfigurations(configurations, services, hosts)
+    items.extend(clusterWideItems)
+    self.validateMinMax(items, recommendedDefaults, configurations)
+    return items
+
+  def getConfigurationsValidationItemsForService(self, configurations, recommendedDefaults, service, services, hosts):
+    items = []
+    serviceName = service["StackServices"]["service_name"]
+    validator = self.validateServiceConfigurations(serviceName)
+    if validator is not None:
+      for siteName, method in validator.items():
+        if siteName in recommendedDefaults:
+          siteProperties = getSiteProperties(configurations, siteName)
+          if siteProperties is not None:
+            siteRecommendations = recommendedDefaults[siteName]["properties"]
+            print("SiteName: %s, method: %s\n" % (siteName, method.__name__))
+            print("Site properties: %s\n" % str(siteProperties))
+            print("Recommendations: %s\n********\n" % str(siteRecommendations))
+            resultItems = method(siteProperties, siteRecommendations, configurations, services, hosts)
+            items.extend(resultItems)
+
+    advisor = self.instantiateServiceAdvisor(service)
+    if advisor is not None:
+      items.extend(advisor.getConfigurationsValidationItems(self, configurations, recommendedDefaults, services, hosts))
+
+    return items
 
   def recommendConfigGroupsConfigurations(self, recommendations, services, components, hosts,
                             servicesList):
@@ -519,10 +691,15 @@ class DefaultStackAdvisor(StackAdvisor):
 
       configurations = {}
 
-      for service in servicesList:
-        calculation = self.getServiceConfigurationRecommender(service)
+      for service in services["services"]:
+        serviceName = service["StackServices"]["service_name"]
+        calculation = self.getServiceConfigurationRecommender(serviceName)
         if calculation is not None:
           calculation(configurations, cgClusterSummary, cgServices, cgHosts)
+        else:
+          advisor = self.instantiateServiceAdvisor(service)
+          if advisor is not None:
+            advisor.getServiceConfigurationRecommendations(self, configuration, cgClusterSummary, cgServices, cgHosts)
 
       cgRecommendation = {
         "configurations": {},
@@ -551,6 +728,8 @@ class DefaultStackAdvisor(StackAdvisor):
                 configElement][property] = value
 
   def recommendConfigurations(self, services, hosts):
+    self.services = services
+
     stackName = services["Versions"]["stack_name"]
     stackVersion = services["Versions"]["stack_version"]
     hostsList = [host["Hosts"]["host_name"] for host in hosts["items"]]
@@ -583,10 +762,15 @@ class DefaultStackAdvisor(StackAdvisor):
     else:
       configurations = recommendations["recommendations"]["blueprint"]["configurations"]
 
-      for service in servicesList:
-        calculation = self.getServiceConfigurationRecommender(service)
+      for service in services["services"]:
+        serviceName = service["StackServices"]["service_name"]
+        calculation = self.getServiceConfigurationRecommender(serviceName)
         if calculation is not None:
           calculation(configurations, clusterSummary, services, hosts)
+        else:
+          advisor = self.instantiateServiceAdvisor(service)
+          if advisor is not None:
+            advisor.getServiceConfigurationRecommendations(self, configurations, clusterSummary, services, hosts)
 
     return recommendations
 
@@ -623,11 +807,19 @@ class DefaultStackAdvisor(StackAdvisor):
 
   def isMasterComponentWithMultipleInstances(self, component):
     componentName = self.getComponentName(component)
+    serviceAdvisor = self.instantiateServiceAdvisorForComponent(componentName)
+    if serviceAdvisor is not None:
+      return serviceAdvisor.isMasterComponentWithMultipleInstances(componentName)
+
     masters = self.getMastersWithMultipleInstances()
     return componentName in masters
 
   def isComponentNotValuable(self, component):
     componentName = self.getComponentName(component)
+    serviceAdvisor = self.instantiateServiceAdvisorForComponent(componentName)
+    if serviceAdvisor is not None:
+      return serviceAdvisor.isComponentNotValuable(componentName)
+
     service = self.getNotValuableComponents()
     return componentName in service
 
@@ -637,6 +829,10 @@ class DefaultStackAdvisor(StackAdvisor):
 
   # Helper dictionaries
   def getComponentCardinality(self, componentName):
+    serviceAdvisor = self.instantiateServiceAdvisorForComponent(componentName)
+    if serviceAdvisor is not None:
+      return serviceAdvisor.getComponentCardinality(componentName)
+
     return self.getCardinalitiesDict().get(componentName, {"min": 1, "max": 1})
 
   def getHostForComponent(self, component, hostsList):
@@ -660,6 +856,10 @@ class DefaultStackAdvisor(StackAdvisor):
     """
     Provides a scheme for laying out given component on different number of hosts.
     """
+    serviceAdvisor = self.instantiateServiceAdvisorForComponent(componentName)
+    if serviceAdvisor is not None:
+      return serviceAdvisor.getComponentLayoutScheme(componentName)
+
     return self.getComponentLayoutSchemes().get(componentName, None)
 
   def getComponentName(self, component):
@@ -667,6 +867,10 @@ class DefaultStackAdvisor(StackAdvisor):
 
   def isComponentNotPreferableOnAmbariServerHost(self, component):
     componentName = self.getComponentName(component)
+    serviceAdvisor = self.instantiateServiceAdvisorForComponent(componentName)
+    if serviceAdvisor is not None:
+      return serviceAdvisor.isComponentNotPreferableOnAmbariServerHost(componentName)
+
     service = self.getNotPreferableOnServerComponents()
     return componentName in service
 
@@ -787,3 +991,9 @@ class DefaultStackAdvisor(StackAdvisor):
       return [int(x) for x in re.sub(r'(\.0+)*$','', v).split(".")]
     return cmp(normalize(version1), normalize(version2))
   pass
+
+def getSiteProperties(configurations, siteName):
+  siteConfig = configurations.get(siteName)
+  if siteConfig is None:
+    return None
+  return siteConfig.get("properties")

ambari-server/src/test/python/stacks/2.3/common/test_stack_advisor.py

@@ -87,7 +87,6 @@ class TestHDP23StackAdvisor(TestCase):
 
 
   def test_createComponentLayoutRecommendations_hawq_1_Host(self):
-    """ Test that HAWQSTANDBY is not recommended on a single node cluster """
 
     services = self.load_json("services-hawq-1-host.json")
     componentsListList = [service["components"] for service in services["services"]]
@@ -99,6 +98,7 @@ class TestHDP23StackAdvisor(TestCase):
     hostsList = [host["Hosts"]["host_name"] for host in hosts["items"]]
     self.assertEquals(len(hostsList), 1)
 
+    self.insertHAWQServiceAdvisorInfo(services)
     recommendations = self.stackAdvisor.createComponentLayoutRecommendations(services, hosts)
 
     recommendedComponentsListList = [hostgroup["components"] for hostgroup in recommendations["blueprint"]["host_groups"]]
@@ -107,6 +107,19 @@ class TestHDP23StackAdvisor(TestCase):
     self.assertFalse('HAWQSTANDBY' in recommendedComponents)
     self.assertTrue('HAWQSEGMENT' in recommendedComponents)
 
+  def insertHAWQServiceAdvisorInfo(self, services):
+    for service in services["services"]:
+      if service["StackServices"]["service_name"] == 'HAWQ':
+        service["StackServices"]["advisor_name"] = "HAWQ200ServiceAdvisor"
+        path = os.path.join(self.testDirectory, '../../../../../main/resources/common-services/HAWQ/2.0.0/service_advisor.py')
+        service["StackServices"]["advisor_path"] = path
+
+  def insertPXFServiceAdvisorInfo(self, services):
+    for service in services["services"]:
+      if service["StackServices"]["service_name"] == 'PXF':
+        service["StackServices"]["advisor_name"] = "PXF300ServiceAdvisor"
+        path = os.path.join(self.testDirectory, '../../../../../main/resources/common-services/PXF/3.0.0/service_advisor.py')
+        service["StackServices"]["advisor_path"] = path
 
   def test_createComponentLayoutRecommendations_hawq_3_Hosts(self):
     """ Test that HAWQSTANDBY is recommended on a 3-node cluster """
@@ -121,9 +134,11 @@ class TestHDP23StackAdvisor(TestCase):
     hostsList = [host["Hosts"]["host_name"] for host in hosts["items"]]
     self.assertEquals(len(hostsList), 3)
 
-    recommendations = self.stackAdvisor.createComponentLayoutRecommendations(services, hosts)
+    self.insertHAWQServiceAdvisorInfo(services)
+    recommendations = self.stackAdvisor.recommendComponentLayout(services, hosts)
+    layoutRecommendations = recommendations["recommendations"]
 
-    recommendedComponentsListList = [hostgroup["components"] for hostgroup in recommendations["blueprint"]["host_groups"]]
+    recommendedComponentsListList = [hostgroup["components"] for hostgroup in layoutRecommendations["blueprint"]["host_groups"]]
     recommendedComponents = [item["name"] for sublist in recommendedComponentsListList for item in sublist]
     self.assertTrue('HAWQMASTER' in recommendedComponents)
     self.assertTrue('HAWQSTANDBY' in recommendedComponents)
@@ -150,9 +165,11 @@ class TestHDP23StackAdvisor(TestCase):
     hostsList = [host["Hosts"]["host_name"] for host in hosts["items"]]
     self.assertEquals(len(hostsList), 3)
 
-    recommendations = self.stackAdvisor.createComponentLayoutRecommendations(services, hosts)
+    self.insertHAWQServiceAdvisorInfo(services)
+    recommendations = self.stackAdvisor.recommendComponentLayout(services, hosts)
+    layoutRecommendations = recommendations["recommendations"]
 
-    recommendedComponentsListList = [hostgroup["components"] for hostgroup in recommendations["blueprint"]["host_groups"]]
+    recommendedComponentsListList = [hostgroup["components"] for hostgroup in layoutRecommendations["blueprint"]["host_groups"]]
     recommendedComponents = [item["name"] for sublist in recommendedComponentsListList for item in sublist]
     self.assertFalse('HAWQMASTER' in recommendedComponents)
     self.assertFalse('HAWQSTANDBY' in recommendedComponents)
@@ -198,6 +215,7 @@ class TestHDP23StackAdvisor(TestCase):
                 }
 
     hawqSegmentHosts = set(["c6401.ambari.apache.org", "c6402.ambari.apache.org", "c6403.ambari.apache.org"])
+    self.insertHAWQServiceAdvisorInfo(services)
     recommendations = self.stackAdvisor.createComponentLayoutRecommendations(services, hosts)
     hostGroups = [ hostgroup["name"] for hostgroup in recommendations["blueprint"]["host_groups"] if {"name": "HAWQSEGMENT"} in hostgroup["components"] ]
     hostNames = [ host["fqdn"] for hostgroup in recommendations["blueprint_cluster_binding"]["host_groups"] if hostgroup["name"] in hostGroups for host in hostgroup["hosts"] ]
@@ -243,6 +261,7 @@ class TestHDP23StackAdvisor(TestCase):
                 }
 
     hawqSegmentHosts = set(["c6401.ambari.apache.org", "c6403.ambari.apache.org"])
+    self.insertHAWQServiceAdvisorInfo(services)
     recommendations = self.stackAdvisor.createComponentLayoutRecommendations(services, hosts)
     hostGroups = [ hostgroup["name"] for hostgroup in recommendations["blueprint"]["host_groups"] if {"name": "HAWQSEGMENT"} in hostgroup["components"] ]
     hostNames = [ host["fqdn"] for hostgroup in recommendations["blueprint_cluster_binding"]["host_groups"] if hostgroup["name"] in hostGroups for host in hostgroup["hosts"] ]
@@ -303,6 +322,7 @@ class TestHDP23StackAdvisor(TestCase):
                 }
 
     hawqSegmentHosts = set(["c6402.ambari.apache.org"])
+    self.insertHAWQServiceAdvisorInfo(services)
     recommendations = self.stackAdvisor.createComponentLayoutRecommendations(services, hosts)
     hostGroups = [ hostgroup["name"] for hostgroup in recommendations["blueprint"]["host_groups"] if {"name": "HAWQSEGMENT"} in hostgroup["components"] ]
     hostNames = [ host["fqdn"] for hostgroup in recommendations["blueprint_cluster_binding"]["host_groups"] if hostgroup["name"] in hostGroups for host in hostgroup["hosts"] ]
@@ -357,6 +377,8 @@ class TestHDP23StackAdvisor(TestCase):
                 }
 
     pxfHosts = set(["c6401.ambari.apache.org", "c6402.ambari.apache.org", "c6403.ambari.apache.org"])
+
+    self.insertPXFServiceAdvisorInfo(services)
     recommendations = self.stackAdvisor.createComponentLayoutRecommendations(services, hosts)
     hostGroups = [ hostgroup["name"] for hostgroup in recommendations["blueprint"]["host_groups"] if {"name": "PXF"} in hostgroup["components"] ]
     hostNames = [ host["fqdn"] for hostgroup in recommendations["blueprint_cluster_binding"]["host_groups"] if hostgroup["name"] in hostGroups for host in hostgroup["hosts"] ]
@@ -411,6 +433,8 @@ class TestHDP23StackAdvisor(TestCase):
                 }
 
     pxfHosts = set(["c6401.ambari.apache.org", "c6402.ambari.apache.org", "c6403.ambari.apache.org"])
+
+    self.insertPXFServiceAdvisorInfo(services)
     recommendations = self.stackAdvisor.createComponentLayoutRecommendations(services, hosts)
     hostGroups = [ hostgroup["name"] for hostgroup in recommendations["blueprint"]["host_groups"] if {"name": "PXF"} in hostgroup["components"] ]
     hostNames = [ host["fqdn"] for hostgroup in recommendations["blueprint_cluster_binding"]["host_groups"] if hostgroup["name"] in hostGroups for host in hostgroup["hosts"] ]
@@ -481,6 +505,8 @@ class TestHDP23StackAdvisor(TestCase):
                 }
 
     pxfHosts = set(["c6402.ambari.apache.org"])
+
+    self.insertPXFServiceAdvisorInfo(services)
     recommendations = self.stackAdvisor.createComponentLayoutRecommendations(services, hosts)
     hostGroups = [ hostgroup["name"] for hostgroup in recommendations["blueprint"]["host_groups"] if {"name": "PXF"} in hostgroup["components"] ]
     hostNames = [ host["fqdn"] for hostgroup in recommendations["blueprint_cluster_binding"]["host_groups"] if hostgroup["name"] in hostGroups for host in hostgroup["hosts"] ]
@@ -507,6 +533,7 @@ class TestHDP23StackAdvisor(TestCase):
     datanodeComponent = self.__getHosts(componentsList, "DATANODE")
     datanodeComponent["hostnames"] = ['c6402.ambari.apache.org']
 
+    self.insertHAWQServiceAdvisorInfo(services)
     validations = self.stackAdvisor.getComponentLayoutValidations(services, hosts)
     expected = {
       'type': 'host-component',
@@ -538,6 +565,7 @@ class TestHDP23StackAdvisor(TestCase):
     hostsList = [host["Hosts"]["host_name"] for host in hosts["items"]]
     self.assertEquals(len(hostsList), 3)
 
+    self.insertPXFServiceAdvisorInfo(services)
     validations = [validation for validation in self.stackAdvisor.getComponentLayoutValidations(services, hosts) if validation["component-name"] == "PXF"]
     self.assertEquals(len(validations), 1)
     expected = {
@@ -566,6 +594,7 @@ class TestHDP23StackAdvisor(TestCase):
     hostsList = [host["Hosts"]["host_name"] for host in hosts["items"]]
     self.assertEquals(len(hostsList), 3)
 
+    self.insertPXFServiceAdvisorInfo(services)
     validations = [validation for validation in self.stackAdvisor.getComponentLayoutValidations(services, hosts) if validation["component-name"] == "PXF"]
     self.assertEquals(len(validations), 1)
     expected = {
@@ -594,6 +623,7 @@ class TestHDP23StackAdvisor(TestCase):
     hostsList = [host["Hosts"]["host_name"] for host in hosts["items"]]
     self.assertEquals(len(hostsList), 3)
 
+    self.insertPXFServiceAdvisorInfo(services)
     validations = [validation for validation in self.stackAdvisor.getComponentLayoutValidations(services, hosts) if validation["component-name"] == "PXF"]
     self.assertEquals(len(validations), 1)
     expected = {
@@ -622,6 +652,7 @@ class TestHDP23StackAdvisor(TestCase):
     hostsList = [host["Hosts"]["host_name"] for host in hosts["items"]]
     self.assertEquals(len(hostsList), 3)
 
+    self.insertPXFServiceAdvisorInfo(services)
     validations = [validation for validation in self.stackAdvisor.getComponentLayoutValidations(services, hosts) if validation["component-name"] == "PXF"]
     self.assertEquals(len(validations), 0)
 
@@ -644,6 +675,7 @@ class TestHDP23StackAdvisor(TestCase):
     hostsList = [host["Hosts"]["host_name"] for host in hosts["items"]]
     self.assertEquals(len(hostsList), 3)
 
+    self.insertHAWQServiceAdvisorInfo(services)
     validations = self.stackAdvisor.getComponentLayoutValidations(services, hosts)
     self.assertEquals(len(validations), 0)
 
@@ -657,6 +689,7 @@ class TestHDP23StackAdvisor(TestCase):
     self.assertEquals(len(hawqStandbyHosts[0]), 1)
     self.assertEquals(hawqMasterHosts[0][0], hawqStandbyHosts[0][0])
 
+    self.insertHAWQServiceAdvisorInfo(services)
     validations = self.stackAdvisor.getComponentLayoutValidations(services, hosts)
     self.assertEquals(len(validations), 1)
     expected = {
@@ -679,6 +712,7 @@ class TestHDP23StackAdvisor(TestCase):
     self.assertNotEquals(hawqMasterHosts[0][0], hawqStandbyHosts[0][0])
     self.assertEquals(hawqMasterHosts[0][0], "c6401.ambari.apache.org")
 
+    self.insertHAWQServiceAdvisorInfo(services)
     validations = self.stackAdvisor.getComponentLayoutValidations(services, hosts)
     self.assertEquals(len(validations), 1)
     expected = {
@@ -703,6 +737,7 @@ class TestHDP23StackAdvisor(TestCase):
     self.assertNotEquals(hawqMasterHosts[0][0], hawqStandbyHosts[0][0])
     self.assertEquals(hawqStandbyHosts[0][0], "c6401.ambari.apache.org")
 
+    self.insertHAWQServiceAdvisorInfo(services)
     validations = self.stackAdvisor.getComponentLayoutValidations(services, hosts)
     self.assertEquals(len(validations), 1)
     expected = {
@@ -734,6 +769,7 @@ class TestHDP23StackAdvisor(TestCase):
     hostsList = [host["Hosts"]["host_name"] for host in hosts["items"]]
     self.assertEquals(len(hostsList), 3)
 
+    self.insertHAWQServiceAdvisorInfo(services)
     validations = self.stackAdvisor.getComponentLayoutValidations(services, hosts)
     self.assertEquals(len(validations), 0)
 
@@ -1939,6 +1975,18 @@ class TestHDP23StackAdvisor(TestCase):
     self.stackAdvisor.recommendTezConfigurations(configurations, clusterData, services, hosts)
     self.assertEquals(configurations, expected)
 
+  def createHAWQServiceAdvisor(self):
+    path = os.path.join(self.testDirectory, '../../../../../main/resources/common-services/HAWQ/2.0.0/service_advisor.py')
+    service = {"service_name" : "HAWQ", "service_version" : "2.0", "advisor_path" : path, "advisor_name" : "HAWQ200ServiceAdvisor"}
+    service = {"StackServices" : service}
+    return self.stackAdvisor.instantiateServiceAdvisor(service)
+
+  def createPXFServiceAdvisor(self):
+    path = os.path.join(self.testDirectory, '../../../../../main/resources/common-services/PXF/3.0.0/service_advisor.py')
+    service = {"service_name" : "PXF", "service_version" : "3.0", "advisor_path" : path, "advisor_name" : "PXF300ServiceAdvisor"}
+    service = {"StackServices" : service}
+    return self.stackAdvisor.instantiateServiceAdvisor(service)
+
   def test_recommendHAWQConfigurations(self):
 
     # original cluster data with 3 segments
@@ -1963,7 +2011,8 @@ class TestHDP23StackAdvisor(TestCase):
 
     # Test 1 - with 3 segments
     self.assertEquals(len(hawqSegmentComponent["hostnames"]), 3)
-    self.stackAdvisor.recommendHAWQConfigurations(configurations, clusterData, services, None)
+    serviceAdvisor = self.createHAWQServiceAdvisor()
+    serviceAdvisor.getServiceConfigurationRecommendations(self.stackAdvisor, configurations, clusterData, services, None)
     self.assertEquals(configurations["hawq-site"]["properties"]["default_hash_table_bucket_number"], str(3 * 6))
     self.assertEquals(configurations["hdfs-client"]["properties"]["output.replace-datanode-on-failure"], "false")
 
@@ -1973,19 +2022,19 @@ class TestHDP23StackAdvisor(TestCase):
 
     # Test 2 - with 100 segments
     hawqSegmentComponent["hostnames"] = ["host" + str(i) for i in range(100)]
-    self.stackAdvisor.recommendHAWQConfigurations(configurations, clusterData, services, None)
+    serviceAdvisor.getServiceConfigurationRecommendations(self.stackAdvisor, configurations, clusterData, services, None)
     self.assertEquals(configurations["hawq-site"]["properties"]["default_hash_table_bucket_number"], str(100 * 5))
     self.assertEquals(configurations["hdfs-client"]["properties"]["output.replace-datanode-on-failure"], "true")
 
     # Test 3 - with 512 segments
     hawqSegmentComponent["hostnames"] = ["host" + str(i) for i in range(512)]
-    self.stackAdvisor.recommendHAWQConfigurations(configurations, clusterData, services, None)
+    serviceAdvisor.getServiceConfigurationRecommendations(self.stackAdvisor, configurations, clusterData, services, None)
     self.assertEquals(configurations["hawq-site"]["properties"]["default_hash_table_bucket_number"], "512")
     self.assertEquals(configurations["hdfs-client"]["properties"]["output.replace-datanode-on-failure"], "true")
 
     # Test 4 - with 513 segments
     hawqSegmentComponent["hostnames"] = ["host" + str(i) for i in range(513)]
-    self.stackAdvisor.recommendHAWQConfigurations(configurations, clusterData, services, None)
+    serviceAdvisor.getServiceConfigurationRecommendations(self.stackAdvisor, configurations, clusterData, services, None)
     self.assertEquals(configurations["hawq-site"]["properties"]["default_hash_table_bucket_number"], "512")
     self.assertEquals(configurations["hdfs-client"]["properties"]["output.replace-datanode-on-failure"], "true")
 
@@ -1993,9 +2042,9 @@ class TestHDP23StackAdvisor(TestCase):
     configurations = {}
     services["configurations"]["hawq-site"] = {"properties":{'hawq-site': {'properties': {}}}}
     hawqSegmentComponent["hostnames"] = []
-    self.stackAdvisor.recommendHAWQConfigurations(configurations, clusterData, services, None)
+    serviceAdvisor.getServiceConfigurationRecommendations(self.stackAdvisor, configurations, clusterData, services, None)
     self.assertEquals(configurations, {'hdfs-client': {'properties': {'output.replace-datanode-on-failure': 'false'}},
-                                       'hawq-site': {'properties': {}}})
+                                       'hawq-site': {'properties': {}},  'hdfs-site': {'properties': {'dfs.allow.truncate': 'true'}}})
 
   def test_validateHiveConfigurations(self):
     properties = {"hive_security_authorization": "None",
@@ -2295,7 +2344,8 @@ class TestHDP23StackAdvisor(TestCase):
       }
     }
 
-    problems = self.stackAdvisor.validateHAWQSiteConfigurations(properties, defaults, configurations, services, hosts)
+    serviceAdvisor = self.createHAWQServiceAdvisor()
+    problems = serviceAdvisor.validateHAWQSiteConfigurations(self.stackAdvisor, properties, defaults, configurations, services, hosts)
     problems_dict = {}
     for problem in problems:
       problems_dict[problem['config-name']] = problem
@@ -2306,7 +2356,7 @@ class TestHDP23StackAdvisor(TestCase):
     configurations["hawq-site"] = {"properties": {"hawq_master_directory": "/data/hawq/master",
                                                   "hawq_segment_directory": "/data/hawq/segment"}}
     properties = configurations["hawq-site"]["properties"]
-    problems = self.stackAdvisor.validateHAWQSiteConfigurations(properties, defaults, configurations, services, hosts)
+    problems = serviceAdvisor.validateHAWQSiteConfigurations(self.stackAdvisor, properties, defaults, configurations, services, hosts)
     problems_dict = {}
     self.assertEqual(len(problems), 0)
     expected_warnings = {}
@@ -2315,7 +2365,7 @@ class TestHDP23StackAdvisor(TestCase):
     configurations["hawq-site"] = {"properties": {"hawq_master_directory": "/data/hawq/master1,/data/hawq/master2",
                                                   "hawq_segment_directory": "/data/hawq/segment1 /data/hawq/segment2"}}
     properties = configurations["hawq-site"]["properties"]
-    problems = self.stackAdvisor.validateHAWQSiteConfigurations(properties, defaults, configurations, services, hosts)
+    problems = serviceAdvisor.validateHAWQSiteConfigurations(self.stackAdvisor, properties, defaults, configurations, services, hosts)
     problems_dict = {}
     for problem in problems:
       problems_dict[problem['config-name']] = problem
@@ -2369,7 +2419,7 @@ class TestHDP23StackAdvisor(TestCase):
                           "level": "ERROR"
                     } ]
     """
-    problems = self.stackAdvisor.validateHAWQSiteConfigurations(properties, defaults, services["configurations"], services, hosts)
+    problems = serviceAdvisor.validateHAWQSiteConfigurations(self.stackAdvisor, properties, defaults, services["configurations"], services, hosts)
     self.assertEqual(len(problems), 1)
     expected = {
       "config-type": "hawq-site",
@@ -2383,20 +2433,20 @@ class TestHDP23StackAdvisor(TestCase):
     # case 2: hawq_global_rm_type is set as yarn, and YARN service is installed. No validation errors expected.
     services["services"].append({"StackServices" : {"service_name" : "YARN"}, "components":[]})
 
-    problems = self.stackAdvisor.validateHAWQSiteConfigurations(properties, defaults, services["configurations"], services, hosts)
+    problems = serviceAdvisor.validateHAWQSiteConfigurations(self.stackAdvisor, properties, defaults, services["configurations"], services, hosts)
     self.assertEqual(len(problems), 0)
 
     # Test HAWQ Master port conflict with Ambari Server Postgres port
 
     # case 1: HAWQ Master is placed on Ambari Server and HAWQ Master port is same as Ambari Server Postgres Port
-    self.stackAdvisor.isHawqMasterComponentOnAmbariServer = MagicMock(return_value=True)
+    serviceAdvisor.isHawqMasterComponentOnAmbariServer = MagicMock(return_value=True)
     configurations = {
       "hawq-site": {
         "properties":
           {"hawq_master_address_port": "5432"}
       }
     }
-    problems = self.stackAdvisor.validateHAWQSiteConfigurations(properties, defaults, configurations, services, hosts)
+    problems = serviceAdvisor.validateHAWQSiteConfigurations(self.stackAdvisor, properties, defaults, configurations, services, hosts)
     self.assertEqual(len(problems), 1)
     expected = {
       "config-name": "hawq_master_address_port",
@@ -2409,21 +2459,21 @@ class TestHDP23StackAdvisor(TestCase):
     self.assertEqual(problems[0], expected)
 
     # case 2: HAWQ Master is placed on Ambari Server and HAWQ Master port is different from  Ambari Server Postgres Port
-    self.stackAdvisor.isHawqMasterComponentOnAmbariServer = MagicMock(return_value=True)
+    serviceAdvisor.isHawqMasterComponentOnAmbariServer = MagicMock(return_value=True)
     configurations["hawq-site"]["properties"]["hawq_master_address_port"] = "10432"
-    problems = self.stackAdvisor.validateHAWQSiteConfigurations(properties, defaults, configurations, services, hosts)
+    problems = serviceAdvisor.validateHAWQSiteConfigurations(self.stackAdvisor, properties, defaults, configurations, services, hosts)
     self.assertEqual(len(problems), 0)
 
     # case 3: HAWQ Master is not placed on Ambari Server and HAWQ Master port is same as  Ambari Server Postgres Port
-    self.stackAdvisor.isHawqMasterComponentOnAmbariServer = MagicMock(return_value=False)
+    serviceAdvisor.isHawqMasterComponentOnAmbariServer = MagicMock(return_value=False)
     configurations["hawq-site"]["properties"]["hawq_master_address_port"] = "5432"
-    problems = self.stackAdvisor.validateHAWQSiteConfigurations(properties, defaults, configurations, services, hosts)
+    problems = serviceAdvisor.validateHAWQSiteConfigurations(self.stackAdvisor, properties, defaults, configurations, services, hosts)
     self.assertEqual(len(problems), 0)
 
     # case 4: HAWQ Master is not placed on Ambari Server and HAWQ Master port is different from  Ambari Server Postgres Port
-    self.stackAdvisor.isHawqMasterComponentOnAmbariServer = MagicMock(return_value=False)
+    serviceAdvisor.isHawqMasterComponentOnAmbariServer = MagicMock(return_value=False)
     configurations["hawq-site"]["properties"]["hawq_master_address_port"] = "10432"
-    problems = self.stackAdvisor.validateHAWQSiteConfigurations(properties, defaults, configurations, services, hosts)
+    problems = serviceAdvisor.validateHAWQSiteConfigurations(self.stackAdvisor, properties, defaults, configurations, services, hosts)
     self.assertEqual(len(problems), 0)
 
     # -------- test query limits warning ----------
@@ -2453,14 +2503,14 @@ class TestHDP23StackAdvisor(TestCase):
       'config-name': 'default_hash_table_bucket_number',
       'level': 'ERROR'
     }
-    problems = self.stackAdvisor.validateHAWQSiteConfigurations(properties, defaults, configurations, services, hosts)
+    problems = serviceAdvisor.validateHAWQSiteConfigurations(self.stackAdvisor, properties, defaults, configurations, services, hosts)
     self.assertEqual(len(problems), 1)
     self.assertEqual(problems[0], expected)
 
     configurations["hawq-site"] = {"properties": {"default_hash_table_bucket_number": "500",
                                                   "hawq_rm_nvseg_perquery_limit": "500"}}
     properties = configurations["hawq-site"]["properties"]
-    problems = self.stackAdvisor.validateHAWQSiteConfigurations(properties, defaults, configurations, services, hosts)
+    problems = serviceAdvisor.validateHAWQSiteConfigurations(self.stackAdvisor, properties, defaults, configurations, services, hosts)
     self.assertEqual(len(problems), 0)
 
 
@@ -2492,19 +2542,20 @@ class TestHDP23StackAdvisor(TestCase):
         'level': 'WARN'
     }
 
-    problems = self.stackAdvisor.validateHAWQHdfsClientConfigurations(properties, defaults, configurations, services, hosts)
+    serviceAdvisor = self.createHAWQServiceAdvisor()
+    problems = serviceAdvisor.validateHAWQHdfsClientConfigurations(self.stackAdvisor, properties, defaults, configurations, services, hosts)
     self.assertEqual(len(problems), 1)
     self.assertEqual(problems[0], expected)
 
     # 2. Try with 3 hosts
     services["services"][0]["components"][0]["StackServiceComponents"]["hostnames"] = ["host1", "host2", "host3"]
-    problems = self.stackAdvisor.validateHAWQHdfsClientConfigurations(properties, defaults, configurations, services, hosts)
+    problems = serviceAdvisor.validateHAWQHdfsClientConfigurations(self.stackAdvisor, properties, defaults, configurations, services, hosts)
     self.assertEqual(len(problems), 1)
     self.assertEqual(problems[0], expected)
 
     # 3. Try with 4 hosts - default value
     services["services"][0]["components"][0]["StackServiceComponents"]["hostnames"] = ["host1", "host2", "host3", "host4"]
-    problems = self.stackAdvisor.validateHAWQHdfsClientConfigurations(properties, defaults, configurations, services, hosts)
+    problems = serviceAdvisor.validateHAWQHdfsClientConfigurations(self.stackAdvisor, properties, defaults, configurations, services, hosts)
     self.assertEqual(len(problems), 0)
 
     # 4. Try with 4 hosts
@@ -2516,7 +2567,7 @@ class TestHDP23StackAdvisor(TestCase):
       'config-name': 'output.replace-datanode-on-failure',
       'level': 'WARN'
     }
-    problems = self.stackAdvisor.validateHAWQHdfsClientConfigurations(properties, defaults, configurations, services, hosts)
+    problems = serviceAdvisor.validateHAWQHdfsClientConfigurations(self.stackAdvisor, properties, defaults, configurations, services, hosts)
     self.assertEqual(len(problems), 1)
     self.assertEqual(problems[0], expected)