@Test(priority = 2) public void testRerunInstance() throws Exception { long currentTime = System.currentTimeMillis(); XJob job = getTestJob("0/10 * * * * ?", queryString, currentTime, currentTime + 180000); LensSessionHandle sessionHandle = scheduler.openSessionAsUser(user); SchedulerJobHandle jobHandle = scheduler.submitAndScheduleJob(sessionHandle, job); Thread.sleep(15000); List<SchedulerJobInstanceInfo> instanceHandleList = scheduler.getSchedulerDAO().getJobInstances(jobHandle); eventService.notifyEvent(mockQueryEnded(instanceHandleList.get(0).getId(), QueryStatus.Status.FAILED)); Thread.sleep(1000); SchedulerJobInstanceInfo info = scheduler.getSchedulerDAO() .getSchedulerJobInstanceInfo(instanceHandleList.get(0).getId()); Assert.assertEquals(info.getInstanceRunList().size(), 1); Assert.assertEquals(info.getInstanceRunList().get(0).getInstanceState(), SchedulerJobInstanceState.FAILED); scheduler.rerunInstance(sessionHandle, instanceHandleList.get(0).getId()); Thread.sleep(5000); eventService.notifyEvent(mockQueryEnded(instanceHandleList.get(0).getId(), QueryStatus.Status.SUCCESSFUL)); Thread.sleep(1000); info = scheduler.getSchedulerDAO().getSchedulerJobInstanceInfo(instanceHandleList.get(0).getId()); Assert.assertEquals(info.getInstanceRunList().size(), 2); Assert.assertEquals(info.getInstanceRunList().get(1).getResultPath(), "/tmp/query1/result"); Assert.assertEquals(info.getInstanceRunList().get(1).getInstanceState(), SchedulerJobInstanceState.SUCCEEDED); scheduler.expireJob(sessionHandle, jobHandle); Assert.assertEquals(scheduler.getSchedulerDAO().getJobState(jobHandle), SchedulerJobState.EXPIRED); scheduler.closeSession(sessionHandle); }
@Override public void rerunInstance(LensSessionHandle sessionHandle, SchedulerJobInstanceHandle instanceHandle) throws LensException { SchedulerJobInstanceInfo instanceInfo = schedulerDAO.getSchedulerJobInstanceInfo(instanceHandle); doesSessionBelongToUser(sessionHandle, schedulerDAO.getUser(instanceInfo.getJobId())); SchedulerJobState currentState = schedulerDAO.getJobState(instanceInfo.getJobId()); if (currentState != SchedulerJobState.SCHEDULED) { throw new LensException(LensSchedulerErrorCode.JOB_IS_NOT_SCHEDULED.getLensErrorInfo(), null, instanceInfo.getJobId().getHandleIdString(), currentState); } List<SchedulerJobInstanceRun> runList = instanceInfo.getInstanceRunList(); if (runList.size() == 0) { throw new LensException(LensSchedulerErrorCode.JOB_INSTANCE_IS_NOT_YET_RUN.getLensErrorInfo(), null, instanceHandle.getHandleIdString(), instanceInfo.getJobId().getHandleIdString()); } SchedulerJobInstanceRun latestRun = runList.get(runList.size() - 1); try { latestRun.getInstanceState().nextTransition(ON_RERUN); notifyRerun(instanceInfo); log.info("Rerunning the instance with {} for job {} ", instanceHandle, instanceInfo.getJobId()); } catch (InvalidStateTransitionException e) { throw new LensException(LensSchedulerErrorCode.INVALID_EVENT_FOR_JOB_INSTANCE.getLensErrorInfo(), e, ON_RERUN.name(), latestRun.getInstanceState().name(), instanceInfo.getId().getHandleIdString(), instanceInfo.getJobId().getHandleIdString()); } }
SchedulerServiceImpl extends BaseLensService implements SchedulerService { @Override public void rerunInstance(LensSessionHandle sessionHandle, SchedulerJobInstanceHandle instanceHandle) throws LensException { SchedulerJobInstanceInfo instanceInfo = schedulerDAO.getSchedulerJobInstanceInfo(instanceHandle); doesSessionBelongToUser(sessionHandle, schedulerDAO.getUser(instanceInfo.getJobId())); SchedulerJobState currentState = schedulerDAO.getJobState(instanceInfo.getJobId()); if (currentState != SchedulerJobState.SCHEDULED) { throw new LensException(LensSchedulerErrorCode.JOB_IS_NOT_SCHEDULED.getLensErrorInfo(), null, instanceInfo.getJobId().getHandleIdString(), currentState); } List<SchedulerJobInstanceRun> runList = instanceInfo.getInstanceRunList(); if (runList.size() == 0) { throw new LensException(LensSchedulerErrorCode.JOB_INSTANCE_IS_NOT_YET_RUN.getLensErrorInfo(), null, instanceHandle.getHandleIdString(), instanceInfo.getJobId().getHandleIdString()); } SchedulerJobInstanceRun latestRun = runList.get(runList.size() - 1); try { latestRun.getInstanceState().nextTransition(ON_RERUN); notifyRerun(instanceInfo); log.info("Rerunning the instance with {} for job {} ", instanceHandle, instanceInfo.getJobId()); } catch (InvalidStateTransitionException e) { throw new LensException(LensSchedulerErrorCode.INVALID_EVENT_FOR_JOB_INSTANCE.getLensErrorInfo(), e, ON_RERUN.name(), latestRun.getInstanceState().name(), instanceInfo.getId().getHandleIdString(), instanceInfo.getJobId().getHandleIdString()); } } }
SchedulerServiceImpl extends BaseLensService implements SchedulerService { @Override public void rerunInstance(LensSessionHandle sessionHandle, SchedulerJobInstanceHandle instanceHandle) throws LensException { SchedulerJobInstanceInfo instanceInfo = schedulerDAO.getSchedulerJobInstanceInfo(instanceHandle); doesSessionBelongToUser(sessionHandle, schedulerDAO.getUser(instanceInfo.getJobId())); SchedulerJobState currentState = schedulerDAO.getJobState(instanceInfo.getJobId()); if (currentState != SchedulerJobState.SCHEDULED) { throw new LensException(LensSchedulerErrorCode.JOB_IS_NOT_SCHEDULED.getLensErrorInfo(), null, instanceInfo.getJobId().getHandleIdString(), currentState); } List<SchedulerJobInstanceRun> runList = instanceInfo.getInstanceRunList(); if (runList.size() == 0) { throw new LensException(LensSchedulerErrorCode.JOB_INSTANCE_IS_NOT_YET_RUN.getLensErrorInfo(), null, instanceHandle.getHandleIdString(), instanceInfo.getJobId().getHandleIdString()); } SchedulerJobInstanceRun latestRun = runList.get(runList.size() - 1); try { latestRun.getInstanceState().nextTransition(ON_RERUN); notifyRerun(instanceInfo); log.info("Rerunning the instance with {} for job {} ", instanceHandle, instanceInfo.getJobId()); } catch (InvalidStateTransitionException e) { throw new LensException(LensSchedulerErrorCode.INVALID_EVENT_FOR_JOB_INSTANCE.getLensErrorInfo(), e, ON_RERUN.name(), latestRun.getInstanceState().name(), instanceInfo.getId().getHandleIdString(), instanceInfo.getJobId().getHandleIdString()); } } SchedulerServiceImpl(CLIService cliService); }
SchedulerServiceImpl extends BaseLensService implements SchedulerService { @Override public void rerunInstance(LensSessionHandle sessionHandle, SchedulerJobInstanceHandle instanceHandle) throws LensException { SchedulerJobInstanceInfo instanceInfo = schedulerDAO.getSchedulerJobInstanceInfo(instanceHandle); doesSessionBelongToUser(sessionHandle, schedulerDAO.getUser(instanceInfo.getJobId())); SchedulerJobState currentState = schedulerDAO.getJobState(instanceInfo.getJobId()); if (currentState != SchedulerJobState.SCHEDULED) { throw new LensException(LensSchedulerErrorCode.JOB_IS_NOT_SCHEDULED.getLensErrorInfo(), null, instanceInfo.getJobId().getHandleIdString(), currentState); } List<SchedulerJobInstanceRun> runList = instanceInfo.getInstanceRunList(); if (runList.size() == 0) { throw new LensException(LensSchedulerErrorCode.JOB_INSTANCE_IS_NOT_YET_RUN.getLensErrorInfo(), null, instanceHandle.getHandleIdString(), instanceInfo.getJobId().getHandleIdString()); } SchedulerJobInstanceRun latestRun = runList.get(runList.size() - 1); try { latestRun.getInstanceState().nextTransition(ON_RERUN); notifyRerun(instanceInfo); log.info("Rerunning the instance with {} for job {} ", instanceHandle, instanceInfo.getJobId()); } catch (InvalidStateTransitionException e) { throw new LensException(LensSchedulerErrorCode.INVALID_EVENT_FOR_JOB_INSTANCE.getLensErrorInfo(), e, ON_RERUN.name(), latestRun.getInstanceState().name(), instanceInfo.getId().getHandleIdString(), instanceInfo.getJobId().getHandleIdString()); } } SchedulerServiceImpl(CLIService cliService); @Override synchronized void init(HiveConf hiveConf); @Override synchronized void start(); @Override void stop(); @Override HealthStatus getHealthStatus(); @Override LensSessionHandle openSessionAsUser(String user); @Override List<SchedulerJobHandle> getAllJobs(String user, SchedulerJobState state, Long start, Long end); @Override SchedulerJobHandle submitJob(LensSessionHandle sessionHandle, XJob job); @Override void scheduleJob(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle); @Override SchedulerJobHandle submitAndScheduleJob(LensSessionHandle sessionHandle, XJob job); @Override XJob getJobDefinition(SchedulerJobHandle jobHandle); @Override SchedulerJobInfo getJobDetails(SchedulerJobHandle jobHandle); @Override void updateJob(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle, XJob newJobDefinition); @Override void expireJob(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle); @Override void suspendJob(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle); @Override void resumeJob(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle); @Override void deleteJob(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle); @Override Collection<SchedulerJobStats> getAllJobStats(String state, String user, String jobName, long startTime,
    long endTime); @Override SchedulerJobStats getJobStats(SchedulerJobHandle handle, String state, long startTime, long endTime); @Override void rerunInstance(LensSessionHandle sessionHandle, SchedulerJobInstanceHandle instanceHandle); @Override List<SchedulerJobInstanceInfo> getJobInstances(SchedulerJobHandle jobHandle, Long numResults); @Override boolean killInstance(LensSessionHandle sessionHandle, SchedulerJobInstanceHandle instanceHandle); @Override SchedulerJobInstanceInfo getInstanceDetails(SchedulerJobInstanceHandle instanceHandle); }
SchedulerServiceImpl extends BaseLensService implements SchedulerService { @Override public void rerunInstance(LensSessionHandle sessionHandle, SchedulerJobInstanceHandle instanceHandle) throws LensException { SchedulerJobInstanceInfo instanceInfo = schedulerDAO.getSchedulerJobInstanceInfo(instanceHandle); doesSessionBelongToUser(sessionHandle, schedulerDAO.getUser(instanceInfo.getJobId())); SchedulerJobState currentState = schedulerDAO.getJobState(instanceInfo.getJobId()); if (currentState != SchedulerJobState.SCHEDULED) { throw new LensException(LensSchedulerErrorCode.JOB_IS_NOT_SCHEDULED.getLensErrorInfo(), null, instanceInfo.getJobId().getHandleIdString(), currentState); } List<SchedulerJobInstanceRun> runList = instanceInfo.getInstanceRunList(); if (runList.size() == 0) { throw new LensException(LensSchedulerErrorCode.JOB_INSTANCE_IS_NOT_YET_RUN.getLensErrorInfo(), null, instanceHandle.getHandleIdString(), instanceInfo.getJobId().getHandleIdString()); } SchedulerJobInstanceRun latestRun = runList.get(runList.size() - 1); try { latestRun.getInstanceState().nextTransition(ON_RERUN); notifyRerun(instanceInfo); log.info("Rerunning the instance with {} for job {} ", instanceHandle, instanceInfo.getJobId()); } catch (InvalidStateTransitionException e) { throw new LensException(LensSchedulerErrorCode.INVALID_EVENT_FOR_JOB_INSTANCE.getLensErrorInfo(), e, ON_RERUN.name(), latestRun.getInstanceState().name(), instanceInfo.getId().getHandleIdString(), instanceInfo.getJobId().getHandleIdString()); } } SchedulerServiceImpl(CLIService cliService); @Override synchronized void init(HiveConf hiveConf); @Override synchronized void start(); @Override void stop(); @Override HealthStatus getHealthStatus(); @Override LensSessionHandle openSessionAsUser(String user); @Override List<SchedulerJobHandle> getAllJobs(String user, SchedulerJobState state, Long start, Long end); @Override SchedulerJobHandle submitJob(LensSessionHandle sessionHandle, XJob job); @Override void scheduleJob(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle); @Override SchedulerJobHandle submitAndScheduleJob(LensSessionHandle sessionHandle, XJob job); @Override XJob getJobDefinition(SchedulerJobHandle jobHandle); @Override SchedulerJobInfo getJobDetails(SchedulerJobHandle jobHandle); @Override void updateJob(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle, XJob newJobDefinition); @Override void expireJob(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle); @Override void suspendJob(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle); @Override void resumeJob(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle); @Override void deleteJob(LensSessionHandle sessionHandle, SchedulerJobHandle jobHandle); @Override Collection<SchedulerJobStats> getAllJobStats(String state, String user, String jobName, long startTime,
    long endTime); @Override SchedulerJobStats getJobStats(SchedulerJobHandle handle, String state, long startTime, long endTime); @Override void rerunInstance(LensSessionHandle sessionHandle, SchedulerJobInstanceHandle instanceHandle); @Override List<SchedulerJobInstanceInfo> getJobInstances(SchedulerJobHandle jobHandle, Long numResults); @Override boolean killInstance(LensSessionHandle sessionHandle, SchedulerJobInstanceHandle instanceHandle); @Override SchedulerJobInstanceInfo getInstanceDetails(SchedulerJobInstanceHandle instanceHandle);  }