@SuppressWarnings("unchecked") @Test(timeout=5000) public void testVertexReRunning() { initDAG(dag); dag.dagScheduler = mock(DAGScheduler.class); startDAG(dag); dispatcher.await(); TezVertexID vId = TezVertexID.getInstance(dagId, 1); Vertex v = dag.getVertex(vId); dispatcher.getEventHandler().handle(new VertexEventTaskCompleted( TezTaskID.getInstance(vId, 0), TaskState.SUCCEEDED)); dispatcher.getEventHandler().handle(new VertexEventTaskCompleted( TezTaskID.getInstance(vId, 1), TaskState.SUCCEEDED)); dispatcher.await(); Assert.assertEquals(VertexState.SUCCEEDED, v.getState()); Assert.assertEquals(1, dag.getSuccessfulVertices()); Assert.assertEquals(1, dag.numCompletedVertices); verify(dag.dagScheduler, times(1)).vertexCompleted(v); dispatcher.getEventHandler().handle( new VertexEventTaskReschedule(TezTaskID.getInstance(vId, 0))); dispatcher.await(); Assert.assertEquals(VertexState.RUNNING, v.getState()); Assert.assertEquals(0, dag.getSuccessfulVertices()); Assert.assertEquals(0, dag.numCompletedVertices); dispatcher.getEventHandler().handle(new VertexEventTaskCompleted( TezTaskID.getInstance(vId, 0), TaskState.SUCCEEDED)); dispatcher.await(); Assert.assertEquals(VertexState.SUCCEEDED, v.getState()); Assert.assertEquals(1, dag.getSuccessfulVertices()); Assert.assertEquals(1, dag.numCompletedVertices); verify(dag.dagScheduler, times(1)).vertexCompleted(v); }
private boolean vertexReRunning(Vertex vertex) { reRunningVertices.add(vertex.getVertexId()); numSuccessfulVertices--; addDiagnostic("Vertex re-running" + ", vertexName=" + vertex.getName() + ", vertexId=" + vertex.getVertexId()); if (!commitAllOutputsOnSuccess) { List<VertexGroupInfo> groupList = vertexGroupInfo.get(vertex.getName()); if (groupList != null) { for (VertexGroupInfo groupInfo : groupList) { if (groupInfo.committed) { LOG.info("Aborting job as committed vertex: " + vertex.getVertexId() + " is re-running"); enactKill(DAGTerminationCause.COMMIT_FAILURE, VertexTerminationCause.COMMIT_FAILURE); return true; } } } } return false; }
DAGImpl implements org.apache.tez.dag.app.dag.DAG,
  EventHandler<DAGEvent> { private boolean vertexReRunning(Vertex vertex) { reRunningVertices.add(vertex.getVertexId()); numSuccessfulVertices--; addDiagnostic("Vertex re-running" + ", vertexName=" + vertex.getName() + ", vertexId=" + vertex.getVertexId()); if (!commitAllOutputsOnSuccess) { List<VertexGroupInfo> groupList = vertexGroupInfo.get(vertex.getName()); if (groupList != null) { for (VertexGroupInfo groupInfo : groupList) { if (groupInfo.committed) { LOG.info("Aborting job as committed vertex: " + vertex.getVertexId() + " is re-running"); enactKill(DAGTerminationCause.COMMIT_FAILURE, VertexTerminationCause.COMMIT_FAILURE); return true; } } } } return false; } }
DAGImpl implements org.apache.tez.dag.app.dag.DAG,
  EventHandler<DAGEvent> { private boolean vertexReRunning(Vertex vertex) { reRunningVertices.add(vertex.getVertexId()); numSuccessfulVertices--; addDiagnostic("Vertex re-running" + ", vertexName=" + vertex.getName() + ", vertexId=" + vertex.getVertexId()); if (!commitAllOutputsOnSuccess) { List<VertexGroupInfo> groupList = vertexGroupInfo.get(vertex.getName()); if (groupList != null) { for (VertexGroupInfo groupInfo : groupList) { if (groupInfo.committed) { LOG.info("Aborting job as committed vertex: " + vertex.getVertexId() + " is re-running"); enactKill(DAGTerminationCause.COMMIT_FAILURE, VertexTerminationCause.COMMIT_FAILURE); return true; } } } } return false; } DAGImpl(TezDAGID dagId,
      Configuration conf,
      DAGPlan jobPlan,
      EventHandler eventHandler,
      TaskAttemptListener taskAttemptListener,
      Credentials dagCredentials,
      Clock clock,
      String appUserName,
      TaskHeartbeatHandler thh,
      AppContext appContext); }
DAGImpl implements org.apache.tez.dag.app.dag.DAG,
  EventHandler<DAGEvent> { private boolean vertexReRunning(Vertex vertex) { reRunningVertices.add(vertex.getVertexId()); numSuccessfulVertices--; addDiagnostic("Vertex re-running" + ", vertexName=" + vertex.getName() + ", vertexId=" + vertex.getVertexId()); if (!commitAllOutputsOnSuccess) { List<VertexGroupInfo> groupList = vertexGroupInfo.get(vertex.getName()); if (groupList != null) { for (VertexGroupInfo groupInfo : groupList) { if (groupInfo.committed) { LOG.info("Aborting job as committed vertex: " + vertex.getVertexId() + " is re-running"); enactKill(DAGTerminationCause.COMMIT_FAILURE, VertexTerminationCause.COMMIT_FAILURE); return true; } } } } return false; } DAGImpl(TezDAGID dagId,
      Configuration conf,
      DAGPlan jobPlan,
      EventHandler eventHandler,
      TaskAttemptListener taskAttemptListener,
      Credentials dagCredentials,
      Clock clock,
      String appUserName,
      TaskHeartbeatHandler thh,
      AppContext appContext); @Override TezDAGID getID(); @Override Configuration getConf(); @Override DAGPlan getJobPlan(); @Override boolean checkAccess(UserGroupInformation callerUGI,
      ApplicationAccessType jobOperation); @Override Vertex getVertex(TezVertexID vertexID); @Override boolean isUber(); @Override Credentials getCredentials(); @Override UserGroupInformation getDagUGI(); @Override DAGState restoreFromEvent(HistoryEvent historyEvent); @Override TezCounters getAllCounters(); static TezCounters incrTaskCounters(
      TezCounters counters, Collection<Vertex> vertices); @Override List<String> getDiagnostics(); @Override DAGReport getReport(); @Override float getProgress(); @Override Map<TezVertexID, Vertex> getVertices(); @Override DAGState getState(); @Override DAGStatusBuilder getDAGStatus(Set<StatusGetOpts> statusOptions); @Override VertexStatusBuilder getVertexStatus(String vertexName,
      Set<StatusGetOpts> statusOptions); @Override /** * The only entry point to change the DAG. */ void handle(DAGEvent event); @Private DAGState getInternalState(); @Override String getUserName(); @Override String getName(); @Override int getTotalVertices(); @Override int getSuccessfulVertices(); @Override Map<ApplicationAccessType, String> getJobACLs(); DAGState initializeDAG(); @Override Vertex getVertex(String vertexName); @Private void constructFinalFullcounters(); @Override boolean isComplete(); }
DAGImpl implements org.apache.tez.dag.app.dag.DAG,
  EventHandler<DAGEvent> { private boolean vertexReRunning(Vertex vertex) { reRunningVertices.add(vertex.getVertexId()); numSuccessfulVertices--; addDiagnostic("Vertex re-running" + ", vertexName=" + vertex.getName() + ", vertexId=" + vertex.getVertexId()); if (!commitAllOutputsOnSuccess) { List<VertexGroupInfo> groupList = vertexGroupInfo.get(vertex.getName()); if (groupList != null) { for (VertexGroupInfo groupInfo : groupList) { if (groupInfo.committed) { LOG.info("Aborting job as committed vertex: " + vertex.getVertexId() + " is re-running"); enactKill(DAGTerminationCause.COMMIT_FAILURE, VertexTerminationCause.COMMIT_FAILURE); return true; } } } } return false; } DAGImpl(TezDAGID dagId,
      Configuration conf,
      DAGPlan jobPlan,
      EventHandler eventHandler,
      TaskAttemptListener taskAttemptListener,
      Credentials dagCredentials,
      Clock clock,
      String appUserName,
      TaskHeartbeatHandler thh,
      AppContext appContext); @Override TezDAGID getID(); @Override Configuration getConf(); @Override DAGPlan getJobPlan(); @Override boolean checkAccess(UserGroupInformation callerUGI,
      ApplicationAccessType jobOperation); @Override Vertex getVertex(TezVertexID vertexID); @Override boolean isUber(); @Override Credentials getCredentials(); @Override UserGroupInformation getDagUGI(); @Override DAGState restoreFromEvent(HistoryEvent historyEvent); @Override TezCounters getAllCounters(); static TezCounters incrTaskCounters(
      TezCounters counters, Collection<Vertex> vertices); @Override List<String> getDiagnostics(); @Override DAGReport getReport(); @Override float getProgress(); @Override Map<TezVertexID, Vertex> getVertices(); @Override DAGState getState(); @Override DAGStatusBuilder getDAGStatus(Set<StatusGetOpts> statusOptions); @Override VertexStatusBuilder getVertexStatus(String vertexName,
      Set<StatusGetOpts> statusOptions); @Override /** * The only entry point to change the DAG. */ void handle(DAGEvent event); @Private DAGState getInternalState(); @Override String getUserName(); @Override String getName(); @Override int getTotalVertices(); @Override int getSuccessfulVertices(); @Override Map<ApplicationAccessType, String> getJobACLs(); DAGState initializeDAG(); @Override Vertex getVertex(String vertexName); @Private void constructFinalFullcounters(); @Override boolean isComplete(); final Configuration conf; }