@Test
public void testRequestFailureBasedOnSuccessFactor() throws Exception {

  ActionQueue aq = new ActionQueue();
  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);
  RequestFactory requestFactory = mock(RequestFactory.class);
  when(fsm.getCluster(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);
  final List<Stage> stages = new ArrayList<Stage>();
  long now = System.currentTimeMillis();
  Stage stage = stageFactory.createNew(1, "/tmp", "cluster1", 1L, "testRequestFailureBasedOnSuccessFactor",
    CLUSTER_HOST_INFO, "", "");
  stage.setStageId(1);
  stage.addHostRoleExecutionCommand("host1", Role.DATANODE, RoleCommand.UPGRADE,
      new ServiceComponentHostUpgradeEvent(Role.DATANODE.toString(), "host1", now, "HDP-0.2"),
      "cluster1", Service.Type.HDFS.toString(), false, false);
  stage.getExecutionCommandWrapper("host1",
      Role.DATANODE.toString()).getExecutionCommand();
  stage.addHostRoleExecutionCommand("host2", Role.DATANODE, RoleCommand.UPGRADE,
      new ServiceComponentHostUpgradeEvent(Role.DATANODE.toString(), "host2", now, "HDP-0.2"),
      "cluster1", Service.Type.HDFS.toString(), false, false);
  stage.getExecutionCommandWrapper("host2",
      Role.DATANODE.toString()).getExecutionCommand();
  stage.addHostRoleExecutionCommand("host3", Role.DATANODE, RoleCommand.UPGRADE,
      new ServiceComponentHostUpgradeEvent(Role.DATANODE.toString(), "host3", now, "HDP-0.2"),
      "cluster1", Service.Type.HDFS.toString(), false, false);
  stage.getExecutionCommandWrapper("host3",
      Role.DATANODE.toString()).getExecutionCommand();
  stages.add(stage);
  stage.getOrderedHostRoleCommands().get(0).setTaskId(1);
  stage.getOrderedHostRoleCommands().get(1).setTaskId(2);
  stage.getOrderedHostRoleCommands().get(2).setTaskId(3);
  stages.add(
      getStageWithSingleTask(
          "host1", "cluster1", Role.HDFS_CLIENT, RoleCommand.UPGRADE, Service.Type.HDFS, 4, 2, 1));
  ActionDBAccessor db = mock(ActionDBAccessor.class);
  HostRoleCommandDAO hostRoleCommandDAOMock = mock(HostRoleCommandDAO.class);
  Mockito.doNothing().when(hostRoleCommandDAOMock).publishTaskCreateEvent(anyListOf(HostRoleCommand.class));
  RequestEntity request = mock(RequestEntity.class);
  when(request.isExclusive()).thenReturn(false);
  when(db.getRequestEntity(anyLong())).thenReturn(request);
  when(db.getCommandsInProgressCount()).thenReturn(stages.size());
  when(db.getStagesInProgress()).thenReturn(stages);
  doAnswer(new Answer<Void>() {
    @Override
    public Void answer(InvocationOnMock invocation) throws Throwable {
      List<CommandReport> reports = (List<CommandReport>) invocation.getArguments()[0];
      for (CommandReport report : reports) {
        String actionId = report.getActionId();
        long[] requestStageIds = StageUtils.getRequestStage(actionId);
        Long requestId = requestStageIds[0];
        Long stageId = requestStageIds[1];
        Long id = report.getTaskId();
        for (Stage stage : stages) {
          if (requestId.equals(stage.getRequestId()) && stageId.equals(stage.getStageId())) {
            for (HostRoleCommand hostRoleCommand : stage.getOrderedHostRoleCommands()) {
              if (hostRoleCommand.getTaskId() == id) {
                hostRoleCommand.setStatus(HostRoleStatus.valueOf(report.getStatus()));
              }
            }
          }
        }
      }
      return null;
    }
  }).when(db).updateHostRoleStates(anyCollectionOf(CommandReport.class));
  when(db.getTask(anyLong())).thenAnswer(new Answer<Object>() {
    @Override
    public Object answer(InvocationOnMock invocation) throws Throwable {
      Long taskId = (Long) invocation.getArguments()[0];
      for (Stage stage : stages) {
        for (HostRoleCommand command : stage.getOrderedHostRoleCommands()) {
          if (taskId.equals(command.getTaskId())) {
            return command;
          }
        }
      }
      return null;
    }
  });
  doAnswer(new Answer<Collection<HostRoleCommandEntity>>() {
    @Override
    public Collection<HostRoleCommandEntity> answer(InvocationOnMock invocation) throws Throwable {
      Long requestId = (Long) invocation.getArguments()[0];
      List<HostRoleCommandEntity> abortedCommands = Lists.newArrayList();
      for (Stage stage : stages) {
        if (requestId.equals(stage.getRequestId())) {
          for (HostRoleCommand command : stage.getOrderedHostRoleCommands()) {
            if (command.getStatus() == HostRoleStatus.QUEUED ||
                command.getStatus() == HostRoleStatus.IN_PROGRESS ||
                command.getStatus() == HostRoleStatus.PENDING) {
              command.setStatus(HostRoleStatus.ABORTED);
              HostRoleCommandEntity hostRoleCommandEntity = command.constructNewPersistenceEntity();
              hostRoleCommandEntity.setStage(stage.constructNewPersistenceEntity());
              abortedCommands.add(hostRoleCommandEntity);
            }
          }
        }
      }
      return abortedCommands;
    }
  }).when(db).abortOperation(anyLong());
  Properties properties = new Properties();
  Configuration conf = new Configuration(properties);
  ActionScheduler scheduler = new ActionScheduler(100, 50, db, aq, fsm, 3,
      new HostsMap((String) null),
      unitOfWork, null, conf, entityManagerProviderMock,
      hostRoleCommandDAOMock, (HostRoleCommandFactory)null);
  ActionManager am = new ActionManager(db, requestFactory, scheduler);
  scheduler.doWork();
  List<CommandReport> reports = new ArrayList<CommandReport>();
  reports.add(getCommandReport(HostRoleStatus.FAILED, Role.DATANODE, Service.Type.HDFS, "1-1", 1));
  am.processTaskResponse("host1", reports, CommandUtils.convertToTaskIdCommandMap(stage.getOrderedHostRoleCommands()));
  reports.clear();
  reports.add(getCommandReport(HostRoleStatus.FAILED, Role.DATANODE, Service.Type.HDFS, "1-1", 2));
  am.processTaskResponse("host2", reports, CommandUtils.convertToTaskIdCommandMap(stage.getOrderedHostRoleCommands()));
  reports.clear();
  reports.add(getCommandReport(HostRoleStatus.COMPLETED, Role.DATANODE, Service.Type.HDFS, "1-1", 3));
  am.processTaskResponse("host3", reports, CommandUtils.convertToTaskIdCommandMap(stage.getOrderedHostRoleCommands()));
  scheduler.doWork();
  Assert.assertEquals(HostRoleStatus.ABORTED, stages.get(1).getHostRoleStatus("host1", "HDFS_CLIENT"));
}