@Test
  public void testIndependentStagesExecution() throws Exception {
    Clusters fsm = mock(Clusters.class);
    Cluster oneClusterMock = mock(Cluster.class);
    Service serviceObj = mock(Service.class);
    ServiceComponent scomp = mock(ServiceComponent.class);
    ServiceComponentHost sch = mock(ServiceComponentHost.class);
    UnitOfWork unitOfWork = mock(UnitOfWork.class);
    AgentCommandsPublisher agentCommandsPublisher = mock(AgentCommandsPublisher.class);
    when(fsm.getCluster(anyString())).thenReturn(oneClusterMock);
    //when(fsm.getHost(anyString())).thenReturn(oneClusterMock);
    when(oneClusterMock.getService(anyString())).thenReturn(serviceObj);
    when(serviceObj.getServiceComponent(anyString())).thenReturn(scomp);
    when(scomp.getServiceComponentHost(anyString())).thenReturn(sch);
    when(serviceObj.getCluster()).thenReturn(oneClusterMock);

    String hostname1 = "ahost.ambari.apache.org";
    String hostname2 = "bhost.ambari.apache.org";
    String hostname3 = "chost.ambari.apache.org";
    String hostname4 = "chost.ambari.apache.org";

    Long hostId1 = 1L;
    Long hostId2 = 2L;
    Long hostId3 = 3L;
    Long hostId4 = 4L;
    HashMap<String, ServiceComponentHost> hosts =
      new HashMap<>();
    hosts.put(hostname1, sch);
    hosts.put(hostname2, sch);
    hosts.put(hostname3, sch);
    hosts.put(hostname4, sch);
    when(scomp.getServiceComponentHosts()).thenReturn(hosts);

    Host host1 = mock(Host.class);
    when(fsm.getHost(hostname1)).thenReturn(host1);
    when(host1.getState()).thenReturn(HostState.HEALTHY);
    when(host1.getHostName()).thenReturn(hostname1);
    when(host1.getHostId()).thenReturn(hostId1);

    Host host2 = mock(Host.class);
    when(fsm.getHost(hostname2)).thenReturn(host2);
    when(host2.getState()).thenReturn(HostState.HEALTHY);
    when(host2.getHostName()).thenReturn(hostname2);
    when(host2.getHostId()).thenReturn(hostId2);

    Host host3 = mock(Host.class);
    when(fsm.getHost(hostname3)).thenReturn(host3);
    when(host3.getState()).thenReturn(HostState.HEALTHY);
    when(host3.getHostName()).thenReturn(hostname3);
    when(host3.getHostId()).thenReturn(hostId3);

    Host host4 = mock(Host.class);
    when(fsm.getHost(hostname4)).thenReturn(host4);
    when(host4.getState()).thenReturn(HostState.HEALTHY);
    when(host4.getHostName()).thenReturn(hostname4);
    when(host4.getHostId()).thenReturn(hostId4);

    List<Stage> firstStageInProgressPerRequest = new ArrayList<>();

    firstStageInProgressPerRequest.add(
            getStageWithSingleTask(
                    hostname1, "cluster1", Role.DATANODE,
                    RoleCommand.START, Service.Type.HDFS, 1, 1, 1));

    // Stage with the same hostname, should not be scheduled
    firstStageInProgressPerRequest.add(
            getStageWithSingleTask(
                    hostname1, "cluster1", Role.GANGLIA_MONITOR,
                    RoleCommand.START, Service.Type.GANGLIA, 2, 2, 2));

    firstStageInProgressPerRequest.add(
            getStageWithSingleTask(
                    hostname2, "cluster1", Role.DATANODE,
                    RoleCommand.START, Service.Type.HDFS, 3, 3, 3));

    firstStageInProgressPerRequest.add(
        getStageWithSingleTask(
            hostname3, "cluster1", Role.DATANODE,
            RoleCommand.START, Service.Type.HDFS, 4, 4, 4));

    ActionDBAccessor db = mock(ActionDBAccessor.class);
    HostRoleCommandDAO hostRoleCommandDAOMock = mock(HostRoleCommandDAO.class);
    Mockito.doNothing().when(hostRoleCommandDAOMock).publishTaskCreateEvent(anyListOf(HostRoleCommand.class));

    List<String> blockingHostsRequest1 = new ArrayList<>();
    when(hostRoleCommandDAOMock.getBlockingHostsForRequest(1, 1)).thenReturn(blockingHostsRequest1);

    List<String> blockingHostsRequest2 = Lists.newArrayList(hostname1);
    when(hostRoleCommandDAOMock.getBlockingHostsForRequest(1, 2)).thenReturn(blockingHostsRequest2);

    RequestEntity request = mock(RequestEntity.class);
    when(request.isExclusive()).thenReturn(false);
    when(request.getClusterHostInfo()).thenReturn(CLUSTER_HOST_INFO);
    when(db.getRequestEntity(anyLong())).thenReturn(request);

    when(db.getCommandsInProgressCount()).thenReturn(firstStageInProgressPerRequest.size());
    when(db.getFirstStageInProgressPerRequest()).thenReturn(firstStageInProgressPerRequest);

    Properties properties = new Properties();
    Configuration conf = new Configuration(properties);
    ActionScheduler scheduler = spy(new ActionScheduler(100, 50, db, fsm, 3,
        new HostsMap((String) null), unitOfWork, null, conf, entityManagerProviderMock,
        hostRoleCommandDAOMock, (HostRoleCommandFactory)null, agentCommandsPublisher));

    doReturn(false).when(scheduler).wasAgentRestartedDuringOperation(any(Host.class), any(Stage.class), anyString());

    scheduler.doWork();

    Assert.assertEquals(HostRoleStatus.QUEUED, firstStageInProgressPerRequest.get(0).getHostRoleStatus(hostname1, "DATANODE"));
    Assert.assertEquals(HostRoleStatus.PENDING, firstStageInProgressPerRequest.get(1).getHostRoleStatus(hostname1, "GANGLIA_MONITOR"));
    Assert.assertEquals(HostRoleStatus.QUEUED, firstStageInProgressPerRequest.get(2).getHostRoleStatus(hostname2, "DATANODE"));
    Assert.assertEquals(HostRoleStatus.QUEUED, firstStageInProgressPerRequest.get(3).getHostRoleStatus(hostname3, "DATANODE"));
  }
Verifies that stages that are executed on different hosts and
rely to different requests are scheduled to be  executed in parallel
ambari-server/src/test/java/org/apache/ambari/server/actionmanager/TestActionScheduler.java