TaskStatusListener { @Subscribe public void onTaskUpdateEvent(TaskUpdateEvent event) { LOG.debug("Received task update event {}", event); List<HostRoleCommand> hostRoleCommandListAll = event.getHostRoleCommands(); List<HostRoleCommand> hostRoleCommandWithReceivedStatus = new ArrayList<>(); Set<StageEntityPK> stagesWithReceivedTaskStatus = new HashSet<>(); Set<Long> requestIdsWithReceivedTaskStatus = new HashSet<>(); for (HostRoleCommand hostRoleCommand : hostRoleCommandListAll) { Long reportedTaskId = hostRoleCommand.getTaskId(); HostRoleCommand activeTask = activeTasksMap.get(reportedTaskId); if (activeTask == null) { LOG.error(String.format("Received update for a task %d which is not being tracked as running task", reportedTaskId)); } else { hostRoleCommandWithReceivedStatus.add(hostRoleCommand); StageEntityPK stageEntityPK = new StageEntityPK(); stageEntityPK.setRequestId(hostRoleCommand.getRequestId()); stageEntityPK.setStageId(hostRoleCommand.getStageId()); stagesWithReceivedTaskStatus.add(stageEntityPK); requestIdsWithReceivedTaskStatus.add(hostRoleCommand.getRequestId()); } } updateActiveTasksMap(hostRoleCommandWithReceivedStatus); Boolean didAnyStageStatusUpdated = updateActiveStagesStatus(stagesWithReceivedTaskStatus, hostRoleCommandListAll); if (didAnyStageStatusUpdated) { updateActiveRequestsStatus(requestIdsWithReceivedTaskStatus, stagesWithReceivedTaskStatus); } } @Inject  TaskStatusListener(TaskEventPublisher taskEventPublisher, StageDAO stageDAO, RequestDAO requestDAO); Map<Long,HostRoleCommand> getActiveTasksMap(); Map<Long, ActiveRequest> getActiveRequestMap(); Map<StageEntityPK, ActiveStage> getActiveStageMap(); @Subscribe void onTaskUpdateEvent(TaskUpdateEvent event); @Subscribe void onTaskCreateEvent(TaskCreateEvent event);  }
@Test public void testOnTaskUpdateEvent() { List<HostRoleCommand> hostRoleCommands = new ArrayList<HostRoleCommand>(); ServiceComponentHostEvent serviceComponentHostEvent = createNiceMock(ServiceComponentHostEvent.class); HostDAO hostDAO = createNiceMock(HostDAO.class); replayAll(); int hostRoleCommandSize = 3; int hrcCounter = 1; for (int stageCounter = 0; stageCounter < 2; stageCounter++) { for (int i = 1; i <= hostRoleCommandSize; i++,hrcCounter++) { String hostname = "hostname-" + hrcCounter; HostRoleCommand hostRoleCommand = new HostRoleCommand(hostname, Role.DATANODE, serviceComponentHostEvent, RoleCommand.EXECUTE, hostDAO, executionCommandDAO, ecwFactory); hostRoleCommand.setStatus(HostRoleStatus.PENDING); hostRoleCommand.setRequestId(1L); hostRoleCommand.setStageId(stageCounter); hostRoleCommand.setTaskId(hrcCounter); hostRoleCommands.add(hostRoleCommand); } } HostRoleStatus hostRoleStatus = HostRoleStatus.PENDING; StageDAO stageDAO = createNiceMock(StageDAO.class); RequestDAO requestDAO = createNiceMock(RequestDAO.class); StageEntity stageEntity = createNiceMock(StageEntity.class); RequestEntity requestEntity = createNiceMock(RequestEntity.class); EasyMock.expect(stageEntity.getStatus()).andReturn(hostRoleStatus).anyTimes();; EasyMock.expect(stageEntity.getDisplayStatus()).andReturn(hostRoleStatus).anyTimes(); EasyMock.expect(stageEntity.isSkippable()).andReturn(Boolean.FALSE).anyTimes();; EasyMock.expect(stageEntity.getRoleSuccessCriterias()).andReturn(Collections.<RoleSuccessCriteriaEntity>emptyList()).anyTimes(); EasyMock.expect(stageDAO.findByPK(anyObject(StageEntityPK.class))).andReturn(stageEntity).anyTimes(); EasyMock.expect(requestEntity.getStatus()).andReturn(hostRoleStatus).anyTimes();; EasyMock.expect(requestEntity.getDisplayStatus()).andReturn(hostRoleStatus).anyTimes(); EasyMock.expect(requestDAO.findByPK(anyLong())).andReturn(requestEntity).anyTimes(); requestDAO.updateStatus(1L,HostRoleStatus.COMPLETED,HostRoleStatus.SKIPPED_FAILED); EasyMock.expectLastCall().times(1); EasyMock.replay(stageEntity); EasyMock.replay(requestEntity); EasyMock.replay(stageDAO); EasyMock.replay(requestDAO); TaskCreateEvent event = new TaskCreateEvent(hostRoleCommands); TaskStatusListener listener = new TaskStatusListener(publisher,stageDAO,requestDAO); Assert.assertTrue(listener.getActiveTasksMap().isEmpty()); Assert.assertTrue(listener.getActiveStageMap().isEmpty()); Assert.assertTrue(listener.getActiveRequestMap().isEmpty()); listener.onTaskCreateEvent(event); Assert.assertEquals(listener.getActiveTasksMap().size(),6); Assert.assertEquals(listener.getActiveStageMap().size(),2); Assert.assertEquals(listener.getActiveRequestMap().size(),1); Assert.assertEquals(listener.getActiveRequestMap().get(1L).getStatus(), hostRoleStatus); String hostname = "hostname-1"; HostRoleCommand hostRoleCommand = new HostRoleCommand(hostname, Role.DATANODE, serviceComponentHostEvent, RoleCommand.EXECUTE, hostDAO, executionCommandDAO, ecwFactory); hostRoleCommand.setStatus(HostRoleStatus.IN_PROGRESS); hostRoleCommand.setRequestId(1L); hostRoleCommand.setStageId(0); hostRoleCommand.setTaskId(1L); listener.onTaskUpdateEvent(new TaskUpdateEvent(Collections.singletonList(hostRoleCommand))); Assert.assertEquals(HostRoleStatus.IN_PROGRESS, listener.getActiveRequestMap().get(1L).getStatus()); hrcCounter = 1; List<HostRoleCommand> finalHostRoleCommands = new ArrayList<HostRoleCommand>(); HostRoleStatus finalHostRoleStatus = HostRoleStatus.COMPLETED; for (int stageCounter = 0; stageCounter < 2; stageCounter++) { for (int i = 1; i <= hostRoleCommandSize; i++,hrcCounter++) { String finalHostname = "hostname-" + hrcCounter; HostRoleCommand finalHostRoleCommand = new HostRoleCommand(finalHostname, Role.DATANODE, serviceComponentHostEvent, RoleCommand.EXECUTE, hostDAO, executionCommandDAO, ecwFactory); finalHostRoleCommand.setStatus(finalHostRoleStatus); finalHostRoleCommand.setRequestId(1L); finalHostRoleCommand.setStageId(stageCounter); finalHostRoleCommand.setTaskId(hrcCounter); finalHostRoleCommands.add(finalHostRoleCommand); } finalHostRoleStatus = HostRoleStatus.SKIPPED_FAILED; } listener.onTaskUpdateEvent(new TaskUpdateEvent(finalHostRoleCommands)); Assert.assertNull(listener.getActiveRequestMap().get(1L)); verifyAll(); }