AMBARI-849. Stage planner implementation.

git-svn-id: https://svn.apache.org/repos/asf/incubator/ambari/branches/AMBARI-666@1398583 13f79535-47bb-0310-9956-ffa450edef68

AMBARI-666-CHANGES.txt

@@ -12,6 +12,8 @@ AMBARI-666 branch (unreleased changes)
 
   NEW FEATURES
 
+  AMBARI-849. Stage planner implementation. (jitendra)
+
   AMBARI-860. Remove code that adds fake stages for testing. (jitendra)
 
   AMBARI-856. Add cluster heatmap. (yusaku)

ambari-server/src/main/java/org/apache/ambari/server/Role.java

@@ -34,5 +34,39 @@ public enum Role {
   HBASE_CLIENT,
   JOBTRACKER,
   TASKTRACKER,
-  MAPREDUCE_CLIENT
+  MAPREDUCE_CLIENT,
+  JAVA_JCE,
+  KERBEROS_SERVER,
+  KERBEROS_CLIENT,
+  KERBEROS_ADMIN_CLIENT,
+  HADOOP_CLIENT,
+  JOBTRACKER_SERVICE_CHECK,
+  MAPREDUCE_SERVICE_CHECK,
+  ZOOKEEPER_SERVICE_CHECK,
+  ZOOKEEPER_QUORUM_SERVICE_CHECK,
+  HBASE_SERVICE_CHECK,
+  MYSQL_SERVER,
+  HIVE_SERVER,
+  HIVE_CLIENT,
+  HIVE_SERVICE_CHECK,
+  HCAT,
+  HCAT_SERVICE_CHECK,
+  OOZIE_CLIENT,
+  OOZIE_SERVER,
+  OOZIE_SERVICE_CHECK,
+  PIG,
+  PIG_SERVICE_CHECK,
+  SQOOP,
+  SQOOP_SERVICE_CHECK,
+  TEMPLETON_CLIENT,
+  TEMPLETON_SERVER,
+  TEMPLETON_SERVICE_CHECK,
+  DASHBOARD,
+  DASHBOARD_SERVICE_CHECK,
+  NAGIOS_SERVER,
+  GANGLIA_SERVER,
+  GANGLIA_MONITOR,
+  GMOND_SERVICE_CHECK,
+  GMETAD_SERVICE_CHECK,
+  MONTOR_WEBSERVER
 }

ambari-server/src/main/java/org/apache/ambari/server/actionmanager/ActionDBAccessor.java

@@ -54,4 +54,12 @@ public interface ActionDBAccessor {
    */
   public long getLastPersistedRequestIdWhenInitialized();
 
+  /**
+   * Updates scheduled stage.
+   * @param s
+   * @param hostname
+   * @param roleStr
+   */
+  public void hostRoleScheduled(Stage s, String hostname, String roleStr);
+
 }

ambari-server/src/main/java/org/apache/ambari/server/actionmanager/ActionDBAccessorImpl.java

@@ -98,4 +98,10 @@ public class ActionDBAccessorImpl implements ActionDBAccessor {
     // TODO Auto-generated method stub
     return 0;
   }
+
+  @Override
+  public void hostRoleScheduled(Stage s, String hostname, String roleStr) {
+    //Update start time, last update time, host role status, number of attempts
+    //in the database.
+  }
 }

ambari-server/src/main/java/org/apache/ambari/server/actionmanager/ActionDBInMemoryImpl.java

@@ -130,4 +130,9 @@ public class ActionDBInMemoryImpl implements ActionDBAccessor {
   public synchronized long getLastPersistedRequestIdWhenInitialized() {
     return lastRequestId;
   }
+  
+  @Override
+  public void hostRoleScheduled(Stage s, String hostname, String roleStr) {
+    //Nothing needed for in-memory implementation
+  }
 }

ambari-server/src/main/java/org/apache/ambari/server/actionmanager/ActionManager.java

@@ -57,6 +57,7 @@ public class ActionManager {
   }
 
   public void start() {
+    LOG.info("Starting scheduler thread");
     scheduler.start();
   }
 

ambari-server/src/main/java/org/apache/ambari/server/actionmanager/ActionScheduler.java

@@ -89,7 +89,7 @@ class ActionScheduler implements Runnable {
   
   private void doWork() throws AmbariException {
     List<Stage> stages = db.getStagesInProgress();
-    LOG.info("Scheduler wakes up, number of stages to look at:"+stages.size());
+    LOG.info("Scheduler wakes up");
     if (stages == null || stages.isEmpty()) {
       //Nothing to do
       LOG.info("No stage in progress..nothing to do");
@@ -164,7 +164,7 @@ class ActionScheduler implements Runnable {
         HostRoleStatus status = s.getHostRoleStatus(host, roleStr);    
         LOG.info("Last attempt time =" + s.getLastAttemptTime(host, roleStr)
             + ", actiontimeout =" + this.actionTimeout + ", current time="
-            + now);
+            + now+", status ="+status+", attempt count="+s.getAttemptCount(host, roleStr));
         if (timeOutActionNeeded(status, s, host, roleStr, now)) {
           LOG.info("Host:" + host + ", role:" + roleStr + ", actionId:"
               + s.getActionId() + " timed out");
@@ -225,9 +225,7 @@ class ActionScheduler implements Runnable {
     long now = System.currentTimeMillis();
     String roleStr = cmd.getRole().toString();
     String hostname = cmd.getHostname();
-    LOG.info("Host:" + hostname + ", role:" + cmd.getRole() + ", actionId:"
-        + s.getActionId() + " being scheduled"+", current time: "+now+", start time: "+
-        s.getStartTime(hostname, roleStr));
+    LOG.info("Going to schedule: "+s.toString());
     if (s.getStartTime(hostname, roleStr) < 0) {
       LOG.info("Update state machine for first attempt");
       try {
@@ -259,6 +257,7 @@ class ActionScheduler implements Runnable {
       throw new AmbariException("Could not get string from jaxb",e);
     }
     actionQueue.enqueue(hostname, cmd);
+    db.hostRoleScheduled(s, hostname, roleStr);
   }
 
   private void updateRoleStats(HostRoleStatus status, RoleStats rs) {

ambari-server/src/main/java/org/apache/ambari/server/actionmanager/HostRoleCommand.java

@@ -121,4 +121,20 @@ class HostRoleCommand {
     HostRoleCommand o = (HostRoleCommand) other;
     return this.role.equals(o.role);
   }
+  
+  @Override
+  public String toString() {
+    StringBuilder builder = new StringBuilder();
+    builder.append("HostRoleCommand State:\n");
+    builder.append("  Role: "+role+"\n");
+    builder.append("  Status: "+status+"\n");
+    builder.append("  Event: "+event+"\n");
+    builder.append("  stdout: "+stdout+"\n");
+    builder.append("  stderr: "+stderr+"\n");
+    builder.append("  exitcode: "+exitCode+"\n");
+    builder.append("  Start time: " + startTime+"\n");
+    builder.append("  Last attempt time: "+lastAttemptTime+"\n");
+    builder.append("  attempt count: "+ attemptCount+"\n");
+    return builder.toString();
+  }
 }

ambari-server/src/main/java/org/apache/ambari/server/actionmanager/Stage.java

@@ -135,6 +135,14 @@ public class Stage {
       return f;
     }
   }
+  
+  public synchronized void setSuccessFactors(Map<Role, Float> suc) {
+    successFactors = suc;
+  }
+  
+  public synchronized Map<Role, Float> getSuccessFactors() {
+    return successFactors;
+  }
 
   public long getRequestId() {
     return requestId;
@@ -227,4 +235,64 @@ public class Stage {
     }
     return false;
   }
+
+  public Map<String, List<ExecutionCommand>> getExecutionCommands() {
+    return this.commandsToSend;
+  }
+
+  public String getLogDir() {
+    return this.logDir;
+  }
+
+  /**
+   * This method should be used only in stage planner. To add
+   * a new execution command use
+   * {@link #addHostRoleExecutionCommand(String, Role, RoleCommand, 
+   * ServiceComponentHostEvent, String, String)}
+   */
+  public synchronized void addExecutionCommand(Stage origStage,
+      ExecutionCommand executionCommand) {
+    String hostname = executionCommand.getHostname();
+    String role = executionCommand.getRole().toString();
+    if (commandsToSend.get(hostname) == null) {
+      commandsToSend.put(hostname, new ArrayList<ExecutionCommand>());
+    }
+    commandsToSend.get(hostname).add(executionCommand);
+    if (hostRoleCommands.get(hostname) == null) {
+      hostRoleCommands.put(hostname, new TreeMap<String, HostRoleCommand>());
+    }
+    hostRoleCommands.get(hostname).put(role,
+        origStage.getHostRoleCommand(hostname, role));
+  }
+
+  private HostRoleCommand getHostRoleCommand(String hostname, String role) {
+    return hostRoleCommands.get(hostname).get(role);
+  }
+  
+  @Override //Object
+  public String toString() {
+    StringBuilder builder = new StringBuilder();
+    builder.append("STAGE DESCRIPTION BEGIN\n");
+    builder.append("requestId="+requestId+"\n");
+    builder.append("stageId="+stageId+"\n");
+    builder.append("clusterName="+clusterName+"\n");
+    builder.append("logDir=" + logDir+"\n");
+    builder.append("Success Factors:\n");
+    for (Role r : successFactors.keySet()) {
+      builder.append("  role: "+r+", factor: "+successFactors.get(r)+"\n");
+    }
+    for (String host : commandsToSend.keySet()) {
+      builder.append("HOST: " + host + " :\n");
+      for (ExecutionCommand ec : commandsToSend.get(host)) {
+        builder.append(ec.toString());
+        builder.append("\n");
+        HostRoleCommand hrc = hostRoleCommands.get(host).get(
+            ec.getRole().toString());
+        builder.append(hrc.toString());
+        builder.append("\n");
+      }
+    }
+    builder.append("STAGE DESCRIPTION END\n");
+    return builder.toString();
+  }
 }

ambari-server/src/main/java/org/apache/ambari/server/agent/ExecutionCommand.java

@@ -23,6 +23,9 @@ import java.util.Map;
 
 import org.apache.ambari.server.Role;
 import org.apache.ambari.server.RoleCommand;
+import org.apache.ambari.server.utils.StageUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.codehaus.jackson.annotate.JsonProperty;
 
 
@@ -31,7 +34,9 @@ import org.codehaus.jackson.annotate.JsonProperty;
  * persisted in the database for recovery.
  */
 public class ExecutionCommand extends AgentCommand {
-
+  
+  private static Log LOG = LogFactory.getLog(ExecutionCommand.class);
+  
   public ExecutionCommand() {
     super(AgentCommandType.EXECUTION_COMMAND);
   }
@@ -71,8 +76,12 @@ public class ExecutionCommand extends AgentCommand {
   
   @Override
   public String toString() {
-    return "Host=" + hostname + ", commandId=" + commandId + ", role=" + role
-        + ", command=" + roleCommand;
+    try {
+      return StageUtils.jaxbToString(this);
+    } catch (Exception ex) {
+      LOG.warn("Exception in json conversion", ex);
+      return "Exception in json conversion"; 
+    }
   }
 
   @Override

ambari-server/src/main/java/org/apache/ambari/server/metadata/RoleCommandOrder.java

@@ -0,0 +1,146 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ambari.server.metadata;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.ambari.server.Role;
+import org.apache.ambari.server.RoleCommand;
+import org.apache.ambari.server.stageplanner.RoleGraphNode;
+
+/**
+ * This class is used to establish the order between two roles. This class
+ * should not be used to determine the dependencies.
+ */
+public class RoleCommandOrder {
+
+  private static class RoleCommandPair {
+    Role role;
+    RoleCommand cmd;
+
+    public RoleCommandPair(Role _role, RoleCommand _cmd) {
+      this.role = _role;
+      this.cmd = _cmd;
+    }
+
+    @Override
+    public int hashCode() {
+      return (role.toString() + cmd.toString()).hashCode();
+    }
+
+    @Override
+    public boolean equals(Object other) {
+      if (other != null && (other instanceof RoleCommandPair)
+          && ((RoleCommandPair) other).role.equals(role)
+          && ((RoleCommandPair) other).cmd.equals(cmd)) {
+        return true;
+      }
+      return false;
+    }
+  }
+
+  /**
+   * key -> blocked role command value -> set of blocker role commands.
+   */
+  private static Map<RoleCommandPair, Set<RoleCommandPair>> dependencies = new HashMap<RoleCommandPair, Set<RoleCommandPair>>();
+
+  private static void addDependency(Role blockedRole,
+      RoleCommand blockedCommand, Role blockerRole, RoleCommand blockerCommand) {
+    RoleCommandPair rcp1 = new RoleCommandPair(blockedRole, blockedCommand);
+    RoleCommandPair rcp2 = new RoleCommandPair(blockerRole, blockerCommand);
+    if (dependencies.get(rcp1) == null) {
+      dependencies.put(rcp1, new HashSet<RoleCommandPair>());
+    }
+    dependencies.get(rcp1).add(rcp2);
+  }
+
+  public static void initialize() {
+    addDependency(Role.SECONDARY_NAMENODE, RoleCommand.START, Role.NAMENODE,
+        RoleCommand.START);
+    addDependency(Role.HBASE_MASTER, RoleCommand.START, Role.ZOOKEEPER_SERVER,
+        RoleCommand.START);
+    addDependency(Role.HBASE_MASTER, RoleCommand.START, Role.NAMENODE,
+        RoleCommand.START);
+    addDependency(Role.HBASE_MASTER, RoleCommand.START, Role.DATANODE,
+        RoleCommand.START);
+    addDependency(Role.HBASE_REGIONSERVER, RoleCommand.START,
+        Role.HBASE_MASTER, RoleCommand.START);
+    addDependency(Role.JOBTRACKER, RoleCommand.START, Role.NAMENODE,
+        RoleCommand.START);
+    addDependency(Role.JOBTRACKER, RoleCommand.START, Role.DATANODE,
+        RoleCommand.START);
+    addDependency(Role.TASKTRACKER, RoleCommand.START, Role.JOBTRACKER,
+        RoleCommand.START);
+    addDependency(Role.OOZIE_SERVER, RoleCommand.START, Role.JOBTRACKER,
+        RoleCommand.START);
+    addDependency(Role.OOZIE_SERVER, RoleCommand.START, Role.TASKTRACKER,
+        RoleCommand.START);
+    addDependency(Role.TEMPLETON_SERVER, RoleCommand.START, Role.TASKTRACKER,
+        RoleCommand.START);
+  }
+
+  /**
+   * Returns the dependency order. -1 => rgn1 before rgn2, 0 => they can be
+   * parallel 1 => rgn2 before rgn1
+   * 
+   * @param roleGraphNode
+   * @param roleGraphNode2
+   */
+  public int order(RoleGraphNode rgn1, RoleGraphNode rgn2) {
+    RoleCommandPair rcp1 = new RoleCommandPair(rgn1.getRole(),
+        rgn1.getCommand());
+    RoleCommandPair rcp2 = new RoleCommandPair(rgn2.getRole(),
+        rgn2.getCommand());
+    if ((dependencies.get(rcp1) != null)
+        && (dependencies.get(rcp1).contains(rcp2))) {
+      return 1;
+    } else if ((dependencies.get(rcp2) != null)
+        && (dependencies.get(rcp2).contains(rcp1))) {
+      return -1;
+    } else if (!rgn2.getCommand().equals(rgn1.getCommand())) {
+      return compareCommands(rgn1.getCommand(), rgn2.getCommand());
+    }
+    return 0;
+  }
+
+  private int compareCommands(RoleCommand rc1, RoleCommand rc2) {
+    if (rc1.equals(rc2)) {
+      throw new IllegalArgumentException("rc1 and rc2 are same");
+    } else if (rc1.equals(RoleCommand.INSTALL)) {
+      return -1;
+    } else if (rc2.equals(RoleCommand.INSTALL)) {
+      return 1;
+    } else if (rc1.equals(RoleCommand.START)) {
+      return -1;
+    } else if (rc2.equals(RoleCommand.START)) {
+      return 1;
+    } else if (rc1.equals(RoleCommand.EXECUTE)) {
+      return -1;
+    } else if (rc2.equals(RoleCommand.EXECUTE)) {
+      return 1;
+    } else if (rc1.equals(RoleCommand.STOP)) {
+      return -1;
+    } else if (rc2.equals(RoleCommand.STOP)) {
+      return 1;
+    }
+    return 0;
+  }
+}

ambari-server/src/main/java/org/apache/ambari/server/stageplanner/RoleGraph.java

@@ -0,0 +1,157 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ambari.server.stageplanner;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+
+import org.apache.ambari.server.actionmanager.Stage;
+import org.apache.ambari.server.agent.ExecutionCommand;
+import org.apache.ambari.server.metadata.RoleCommandOrder;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+public class RoleGraph {
+  
+  private static Log LOG = LogFactory.getLog(RoleGraph.class);
+  
+  Map<String, RoleGraphNode> graph = null;
+  private RoleCommandOrder roleDependencies;
+  private Stage initialStage = null;
+  private boolean sameHostOptimization = true;
+ 
+  public RoleGraph(RoleCommandOrder rd) {
+    this.roleDependencies = rd;
+  }
+  
+  /**
+   * Given a stage builds a DAG of all execution commands within the stage.
+   */
+  public void build(Stage stage) {
+    if (stage == null) {
+      throw new IllegalArgumentException("Null stage");
+    }
+    graph = new TreeMap<String, RoleGraphNode>();
+    initialStage = stage;
+    Map<String, List<ExecutionCommand>> execCommands = stage
+        .getExecutionCommands();
+    for (String host : execCommands.keySet()) {
+      for (ExecutionCommand exec : execCommands.get(host)) {
+        String role = exec.getRole().toString();
+        RoleGraphNode rgn;
+        if (graph.get(role) == null) {
+           rgn = new RoleGraphNode(exec.getRole(),
+              exec.getRoleCommand());
+          graph.put(role, rgn);
+        }
+        rgn = graph.get(role);
+        rgn.addHost(host);
+      }
+    }
+    //Add edges
+    for (String roleI : graph.keySet()) {
+      for (String roleJ : graph.keySet()) {
+        if (roleI.equals(roleJ)) {
+          continue;
+        } else {
+          RoleGraphNode rgnI = graph.get(roleI);
+          RoleGraphNode rgnJ = graph.get(roleJ);
+          int order = roleDependencies.order(rgnI, rgnJ);
+          if (order == -1) {
+            rgnI.addEdge(rgnJ);
+          } else if (order == 1) {
+            rgnJ.addEdge(rgnI);
+          }
+        }
+      }
+    }
+  }
+  
+  /**
+   * Returns a list of stages that need to be executed one after another
+   * to execute the DAG generated in the last {@link #build(Stage)} call.
+   */
+  public List<Stage> getStages() {
+    long initialStageId = initialStage.getStageId();
+    List<Stage> stageList = new ArrayList<Stage>();
+    List<RoleGraphNode> firstStageNodes = new ArrayList<RoleGraphNode>();
+    while (!graph.isEmpty()) {
+      LOG.info(this.stringifyGraph());
+
+      for (String role: graph.keySet()) {
+        RoleGraphNode rgn = graph.get(role);
+        if (rgn.getInDegree() == 0) {
+          firstStageNodes.add(rgn);
+        }
+      }
+      Stage aStage = getStageFromGraphNodes(initialStage, firstStageNodes);
+      aStage.setStageId(initialStageId++);
+      stageList.add(aStage);
+      //Remove first stage nodes from the graph, we know that none of
+      //these nodes have an incoming edges.
+      for (RoleGraphNode rgn : firstStageNodes) {
+        if (this.sameHostOptimization) {
+          //Perform optimization
+        }
+        removeZeroInDegreeNode(rgn.getRole().toString());
+      }
+      firstStageNodes.clear();
+    }
+    return stageList;
+  }
+
+  /**
+   * Assumes there are no incoming edges.
+   */
+  private synchronized void removeZeroInDegreeNode(String role) {
+    RoleGraphNode nodeToRemove = graph.remove(role);
+    for (RoleGraphNode edgeNode: nodeToRemove.getEdges()) {
+      edgeNode.decrementInDegree();
+    }
+  }
+
+  private Stage getStageFromGraphNodes(Stage origStage,
+      List<RoleGraphNode> stageGraphNodes) {
+    Stage newStage = new Stage(origStage.getRequestId(),
+        origStage.getLogDir(), origStage.getClusterName());
+    newStage.setSuccessFactors(origStage.getSuccessFactors());
+    for (RoleGraphNode rgn : stageGraphNodes) {
+      for (String host : rgn.getHosts()) {
+        newStage.addExecutionCommand(origStage,
+            origStage.getExecutionCommand(host, rgn.getRole().toString()));
+      }
+    }
+    return newStage;
+  }
+  
+  public String stringifyGraph() {
+    StringBuilder builder = new StringBuilder();
+    builder.append("Graph:\n");
+    for (String role : graph.keySet()) {
+      builder.append(graph.get(role));
+      for (RoleGraphNode rgn : graph.get(role).getEdges()) {
+        builder.append(" --> ");
+        builder.append(rgn);
+      }
+      builder.append("\n");
+    }
+    return builder.toString();
+  }
+}

ambari-server/src/main/java/org/apache/ambari/server/stageplanner/RoleGraphNode.java

@@ -0,0 +1,78 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ambari.server.stageplanner;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+
+import org.apache.ambari.server.Role;
+import org.apache.ambari.server.RoleCommand;
+
+public class RoleGraphNode {
+  public RoleGraphNode(Role role, RoleCommand command) {
+    this.role = role;
+    this.command = command;
+  }
+  private Role role;
+  private RoleCommand command;
+  private int inDegree = 0;
+  private List<String> hosts = new ArrayList<String>();
+  private Map<String, RoleGraphNode> edges = new TreeMap<String, RoleGraphNode>();
+  public synchronized void addHost(String host) {
+    hosts.add(host);
+  }
+  public synchronized void addEdge(RoleGraphNode rgn) {
+    if (edges.containsKey(rgn.getRole().toString())) {
+      return;
+    }
+    edges.put(rgn.getRole().toString(), rgn);
+    rgn.incrementInDegree();
+  }
+  private synchronized void incrementInDegree() {
+    inDegree ++;
+  }
+  public Role getRole() {
+    return role;
+  }
+  public RoleCommand getCommand() {
+    return command;
+  }
+  public List<String> getHosts() {
+    return hosts;
+  }
+  public int getInDegree() {
+    return inDegree;
+  }
+  
+  Collection<RoleGraphNode> getEdges() {
+    return edges.values();
+  }
+  public synchronized void decrementInDegree() {
+    inDegree --;
+  }
+  
+  @Override
+  public String toString() {
+    StringBuilder builder = new StringBuilder();
+    builder.append("("+role+", "+command +", "+inDegree+")");
+    return builder.toString();
+  }
+}

ambari-server/src/main/java/org/apache/ambari/server/utils/StageUtils.java

@@ -56,15 +56,18 @@ public class StageUtils {
     return requestStageIds;
   }
 
+  public static Stage getATestStage(long requestId, long stageId) {
+    String hostname;
+    try {
+      hostname = InetAddress.getLocalHost().getHostName();
+    } catch (UnknownHostException e) {
+      hostname = "host-dummy";
+    }
+    return getATestStage(requestId, stageId, hostname);
+  }
+  
   //For testing only
   public static Stage getATestStage(long requestId, long stageId, String hostname) {
-    if (hostname == null || "".equals(hostname)) {
-      try {
-        hostname = InetAddress.getLocalHost().getHostName();
-      } catch (UnknownHostException e) {
-        hostname = "host-dummy";
-      }
-    }
     Stage s = new Stage(requestId, "/tmp", "cluster1");
     s.setStageId(stageId);
     long now = System.currentTimeMillis();
@@ -90,11 +93,6 @@ public class StageUtils {
     Map<String, String> roleParams = new TreeMap<String, String>();
     roleParams.put("format", "false");
     execCmd.setRoleParams(roleParams);
-    try {
-      LOG.info("Command string = " + StageUtils.jaxbToString(execCmd));
-    } catch (Exception e) {
-      throw new RuntimeException("Could not get string from jaxb",e);
-    }
     return s;
   }
   

ambari-server/src/test/java/org/apache/ambari/server/actionmanager/TestActionScheduler.java

@@ -28,6 +28,8 @@ import static org.mockito.Mockito.when;
 import java.util.ArrayList;
 import java.util.List;
 
+import junit.framework.Assert;
+
 import org.apache.ambari.server.Role;
 import org.apache.ambari.server.agent.ActionQueue;
 import org.apache.ambari.server.agent.AgentCommand;
@@ -69,20 +71,17 @@ public class TestActionScheduler {
 
     //Keep large number of attempts so that the task is not expired finally
     //Small action timeout to test rescheduling
-    ActionScheduler scheduler = new ActionScheduler(100, 100, db, aq, fsm, 10000);
-    // Start the thread
+    ActionScheduler scheduler = new ActionScheduler(100, 100, db, aq, fsm,
+        10000);
+   // Start the thread
     scheduler.start();
 
-    Thread.sleep(200);
-    List<AgentCommand> ac = aq.dequeueAll(hostname);
-    assertEquals(1, ac.size());
+    List<AgentCommand> ac = waitForQueueSize(hostname, aq, 1);
     assertTrue(ac.get(0) instanceof ExecutionCommand);
     assertEquals("1-977", ((ExecutionCommand) (ac.get(0))).getCommandId());
 
     //The action status has not changed, it should be queued again.
-    Thread.sleep(200);
-    ac = aq.dequeueAll(hostname);
-    assertEquals(1, ac.size());
+    ac = waitForQueueSize(hostname, aq, 1);
     assertTrue(ac.get(0) instanceof ExecutionCommand);
     assertEquals("1-977", ((ExecutionCommand) (ac.get(0))).getCommandId());
 
@@ -91,9 +90,24 @@ public class TestActionScheduler {
     ac = aq.dequeueAll(hostname);
 
     //Wait for sometime, it shouldn't be scheduled this time.
-    Thread.sleep(200);
-    ac = aq.dequeueAll(hostname);
-    assertEquals(0, ac.size());
+    ac = waitForQueueSize(hostname, aq, 0);
+    scheduler.stop();
+  }
+
+  private List<AgentCommand> waitForQueueSize(String hostname, ActionQueue aq,
+      int expectedQueueSize) throws InterruptedException {
+    while (true) {
+      List<AgentCommand> ac = aq.dequeueAll(hostname);
+      if (ac != null) {
+        if (ac.size() == expectedQueueSize) {
+          return ac;
+        } else if (ac.size() > expectedQueueSize) {
+          Assert.fail("Expected size : " + expectedQueueSize + " Actual size="
+              + ac.size());
+        }
+      }
+      Thread.sleep(100);
+    }
   }
 
   /**
@@ -113,22 +127,23 @@ public class TestActionScheduler {
     when(scomp.getServiceComponentHost(anyString())).thenReturn(sch);
     when(serviceObj.getCluster()).thenReturn(oneClusterMock);
 
-    ActionDBAccessorImpl db = mock(ActionDBAccessorImpl.class);
+    ActionDBAccessor db = new ActionDBInMemoryImpl();
     String hostname = "ahost.ambari.apache.org";
     List<Stage> stages = new ArrayList<Stage>();
     Stage s = StageUtils.getATestStage(1, 977, hostname);
     stages.add(s);
-    when(db.getStagesInProgress()).thenReturn(stages);
+    db.persistActions(stages);
 
-    //Keep large number of attempts so that the task is not expired finally
     //Small action timeout to test rescheduling
-    ActionScheduler scheduler = new ActionScheduler(100, 100, db, aq, fsm, 3);
+    ActionScheduler scheduler = new ActionScheduler(100, 50, db, aq, fsm, 3);
     // Start the thread
     scheduler.start();
 
-    Thread.sleep(500);
-    //TODO timeoutHostRole must be called exactly once but in this case the state
-    //in the db continues to be pending therefore it is processed multiple times.
-    verify(db, atLeastOnce()).timeoutHostRole(hostname, 1, 977, Role.NAMENODE);
+    while (!stages.get(0).getHostRoleStatus(hostname, "NAMENODE")
+        .equals(HostRoleStatus.TIMEDOUT)) {
+      Thread.sleep(100);
+    }
+    assertEquals(stages.get(0).getHostRoleStatus(hostname, "NAMENODE"),
+        HostRoleStatus.TIMEDOUT);
   }
 }

ambari-server/src/test/java/org/apache/ambari/server/stageplanner/TestStagePlanner.java

@@ -0,0 +1,128 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ambari.server.stageplanner;
+
+import static org.junit.Assert.*;
+
+import java.util.List;
+
+import org.apache.ambari.server.Role;
+import org.apache.ambari.server.RoleCommand;
+import org.apache.ambari.server.actionmanager.Stage;
+import org.apache.ambari.server.metadata.RoleCommandOrder;
+import org.apache.ambari.server.state.svccomphost.ServiceComponentHostInstallEvent;
+import org.apache.ambari.server.state.svccomphost.ServiceComponentHostStartEvent;
+import org.apache.ambari.server.utils.StageUtils;
+import org.junit.Test;
+
+public class TestStagePlanner {
+  
+  @Test
+  public void testSingleStagePlan() {
+    RoleCommandOrder.initialize();
+    RoleCommandOrder rco = new RoleCommandOrder();
+    RoleGraph rg = new RoleGraph(rco);
+    String hostname = "dummy";
+    Stage stage = StageUtils.getATestStage(1, 1, hostname);
+    rg.build(stage);
+    List<Stage> outStages = rg.getStages();
+    for (Stage s: outStages) {
+      System.out.println(s.toString());
+    }
+    assertEquals(1, outStages.size());
+    assertEquals(stage.getExecutionCommands(hostname), outStages.get(0)
+        .getExecutionCommands(hostname));
+  }
+  
+  @Test
+  public void testMultiStagePlan() {
+    RoleCommandOrder.initialize();
+    RoleCommandOrder rco = new RoleCommandOrder();
+    RoleGraph rg = new RoleGraph(rco);
+    long now = System.currentTimeMillis();
+    Stage stage = StageUtils.getATestStage(1, 1, "host1");
+    stage.addHostRoleExecutionCommand("host2", Role.HBASE_MASTER,
+        RoleCommand.START, new ServiceComponentHostStartEvent("HBASE_MASTER",
+            "host2", now), "cluster1", "HBASE");
+    stage.addHostRoleExecutionCommand("host3", Role.ZOOKEEPER_SERVER,
+        RoleCommand.START, new ServiceComponentHostStartEvent("ZOOKEEPER_SERVER",
+            "host3", now), "cluster1", "ZOOKEEPER");
+    System.out.println(stage.toString());
+    
+    rg.build(stage);
+    System.out.println(rg.stringifyGraph());
+    List<Stage> outStages = rg.getStages();
+    for (Stage s: outStages) {
+      System.out.println(s.toString());
+    }
+    assertEquals(3, outStages.size());
+  }
+  
+  @Test
+  public void testManyStages() {
+    RoleCommandOrder.initialize();
+    RoleCommandOrder rco = new RoleCommandOrder();
+    RoleGraph rg = new RoleGraph(rco);
+    long now = System.currentTimeMillis();
+    Stage stage = StageUtils.getATestStage(1, 1, "host1");
+    stage.addHostRoleExecutionCommand("host11", Role.SECONDARY_NAMENODE,
+        RoleCommand.START, new ServiceComponentHostStartEvent("SECONDARY_NAMENODE",
+            "host11", now), "cluster1", "HDFS");
+    stage.addHostRoleExecutionCommand("host2", Role.HBASE_MASTER,
+        RoleCommand.START, new ServiceComponentHostStartEvent("HBASE_MASTER",
+            "host2", now), "cluster1", "HBASE");
+    stage.addHostRoleExecutionCommand("host3", Role.ZOOKEEPER_SERVER,
+        RoleCommand.START, new ServiceComponentHostStartEvent("ZOOKEEPER_SERVER",
+            "host3", now), "cluster1", "ZOOKEEPER");
+    stage.addHostRoleExecutionCommand("host4", Role.DATANODE,
+        RoleCommand.START, new ServiceComponentHostStartEvent("DATANODE",
+            "host4", now), "cluster1", "HDFS");
+    stage.addHostRoleExecutionCommand("host4", Role.HBASE_REGIONSERVER,
+        RoleCommand.START, new ServiceComponentHostStartEvent("HBASE_REGIONSERVER",
+            "host4", now), "cluster1", "HBASE");
+    stage.addHostRoleExecutionCommand("host4", Role.TASKTRACKER,
+        RoleCommand.START, new ServiceComponentHostStartEvent("TASKTRACKER",
+            "host4", now), "cluster1", "MAPREDUCE");
+    stage.addHostRoleExecutionCommand("host5", Role.JOBTRACKER,
+        RoleCommand.START, new ServiceComponentHostStartEvent("JOBTRACKER",
+            "host5", now), "cluster1", "MAPREDUCE");
+    stage.addHostRoleExecutionCommand("host6", Role.OOZIE_SERVER,
+        RoleCommand.START, new ServiceComponentHostStartEvent("OOZIE_SERVER",
+            "host6", now), "cluster1", "OOZIE");
+    stage.addHostRoleExecutionCommand("host7", Role.TEMPLETON_SERVER,
+        RoleCommand.START, new ServiceComponentHostStartEvent("TEMPLETON_SERVER",
+            "host7", now), "cluster1", "TEMPLETON");
+    stage.addHostRoleExecutionCommand("host8", Role.NAGIOS_SERVER,
+        RoleCommand.START, new ServiceComponentHostStartEvent("NAGIOS_SERVER",
+            "host8", now), "cluster1", "NAGIOS");
+    stage.addHostRoleExecutionCommand("host4", Role.GANGLIA_MONITOR,
+        RoleCommand.START, new ServiceComponentHostStartEvent("GANGLIA_MONITOR",
+            "host4", now), "cluster1", "GANGLIA");
+    stage.addHostRoleExecutionCommand("host9", Role.GANGLIA_SERVER,
+        RoleCommand.START, new ServiceComponentHostStartEvent("GANGLIA_SERVER",
+            "host9", now), "cluster1", "GANGLIA");
+    System.out.println(stage.toString());
+    rg.build(stage);
+    System.out.println(rg.stringifyGraph());
+    List<Stage> outStages = rg.getStages();
+    for (Stage s: outStages) {
+      System.out.println(s.toString());
+    }
+    assertEquals(5, outStages.size());
+  }
+}