InstanceRunning extends InstanceState { @Override public State run(InstanceStateContext context) throws OdpsException { try { reporter = new ReporterThread(context); reporter.setDaemon(true); reporter.start(); } catch (Exception e) { } if (context.getExecutionContext().isInteractiveMode()) { SignalHandler instanceRunningIntSignalHandler = SignalUtil.getInstanceRunningIntSignalHandler(Thread.currentThread()); SignalUtil.registerSignalHandler("INT", instanceRunningIntSignalHandler); } try { polling(context); } finally { if (context.getExecutionContext().isInteractiveMode()) { SignalHandler defaultIntSignalHandler = SignalUtil.getDefaultIntSignalHandler(Thread.currentThread()); SignalUtil.registerSignalHandler("INT", defaultIntSignalHandler); } stopReporter(); } return new InstanceTerminated(); }  @Override State run(InstanceStateContext context);  }
@Test public void testInstanceRunning_GetProgressFail() throws Exception { ExecutionContext context = ExecutionContext.init(); Odps odps = Mockito.spy(OdpsConnectionFactory.createOdps(context)); Instances instances = Mockito.spy(odps.instances()); SQLTask task = new SQLTask(); String taskname = "sqltest"; task.setQuery("select count(*) from src;"); task.setName(taskname); Instance instance = Mockito.spy(instances.create(task)); statusCount = 0; progressCount = 0; doAnswer(new Answer() { @Override public Object answer(InvocationOnMock invocation) throws Throwable { statusCount ++; if (statusCount > 15) { return Instance.Status.TERMINATED; } return Instance.Status.RUNNING; } }).when(instance).getStatus(true); doAnswer(new Answer() { @Override public Object answer(InvocationOnMock invocation) throws Throwable { progressCount ++; if (progressCount == 1) { return null; } if (progressCount == 2) { List<Instance.StageProgress> res = new ArrayList<Instance.StageProgress>(); return res; } if (progressCount == 3 || progressCount == 4) { List<Instance.StageProgress> res = new ArrayList<Instance.StageProgress>(); res.add(progress); return res; } if (progressCount == 5) { List<Instance.StageProgress> res = new ArrayList<Instance.StageProgress>(); res.add(progress); res.add(progress); return res; } if (progressCount > 6) { List<Instance.StageProgress> res = new ArrayList<Instance.StageProgress>(); res.add(progress); return res; } throw new OdpsException("get progress failed."); } }).when(instance).getTaskProgress(taskname); InstanceStateContext cont = Mockito.spy(new InstanceStateContext(odps, instance, context)); InstanceRunning running = Mockito.spy(new InstanceRunning()); running.run(cont); Assert.assertEquals(1, cont.getTaskProgress().size()); }