@Test(timeout = 1000) @SuppressWarnings("unchecked") public void testDAGClientHandler() throws TezException { TezDAGID mockTezDAGId = mock(TezDAGID.class); when(mockTezDAGId.getId()).thenReturn(1); when(mockTezDAGId.toString()).thenReturn("dag_9999_0001_1"); DAG mockDAG = mock(DAG.class); when(mockDAG.getID()).thenReturn(mockTezDAGId); DAGStatusBuilder mockDagStatusBuilder = mock(DAGStatusBuilder.class); when(mockDAG.getDAGStatus(anySetOf(StatusGetOpts.class))).thenReturn( mockDagStatusBuilder); VertexStatusBuilder mockVertexStatusBuilder = mock(VertexStatusBuilder.class); when(mockDAG.getVertexStatus(anyString(), anySetOf(StatusGetOpts.class))) .thenReturn(mockVertexStatusBuilder); DAGAppMaster mockDagAM = mock(DAGAppMaster.class); AppContext mockAppContext = mock(AppContext.class); when(mockDagAM.getContext()).thenReturn(mockAppContext); when(mockDagAM.getContext().getCurrentDAG()).thenReturn(mockDAG); DAGClientHandler dagClientHandler = new DAGClientHandler(mockDagAM); assertEquals(1, dagClientHandler.getAllDAGs().size()); assertEquals("dag_9999_0001_1", dagClientHandler.getAllDAGs().get(0)); try { dagClientHandler.getDAGStatus("dag_9999_0001_2", Sets.newSet(StatusGetOpts.GET_COUNTERS)); fail("should not come here"); } catch (TezException e) { assertTrue(e.getMessage().contains("Unknown dagId")); } DAGStatus dagStatus = dagClientHandler.getDAGStatus("dag_9999_0001_1", Sets.newSet(StatusGetOpts.GET_COUNTERS)); assertEquals(mockDagStatusBuilder, dagStatus); try { dagClientHandler.getVertexStatus("dag_9999_0001_2", "v1", Sets.newSet(StatusGetOpts.GET_COUNTERS)); fail("should not come here"); } catch (TezException e) { assertTrue(e.getMessage().contains("Unknown dagId")); } VertexStatus vertexStatus = dagClientHandler.getVertexStatus("dag_9999_0001_1", "v1", Sets.newSet(StatusGetOpts.GET_COUNTERS)); assertEquals(mockVertexStatusBuilder, vertexStatus); when(mockDagAM.isSession()).thenReturn(false); try{ dagClientHandler.getSessionStatus(); fail("should not come here"); }catch(TezException e){ assertEquals("Unsupported operation as AM not running in session mode", e.getMessage()); } when(mockDagAM.isSession()).thenReturn(true); when(mockDagAM.getState()).thenReturn(DAGAppMasterState.INITED); assertEquals(TezAppMasterStatus.INITIALIZING, dagClientHandler.getSessionStatus()); when(mockDagAM.getState()).thenReturn(DAGAppMasterState.ERROR); assertEquals(TezAppMasterStatus.SHUTDOWN, dagClientHandler.getSessionStatus()); PreWarmContext mockPreWarnContext = mock(PreWarmContext.class); dagClientHandler.preWarmContainers(mockPreWarnContext); verify(mockDagAM).startPreWarmContainers(mockPreWarnContext); try{ dagClientHandler.tryKillDAG("dag_9999_0001_2"); fail("should not come here"); }catch(TezException e){ assertTrue(e.getMessage().contains("Unknown dagId")); } dagClientHandler.tryKillDAG("dag_9999_0001_1"); ArgumentCaptor<DAG> eventCaptor = ArgumentCaptor.forClass(DAG.class); verify(mockDagAM, times(1)).tryKillDAG(eventCaptor.capture()); assertEquals(1, eventCaptor.getAllValues().size()); assertTrue(eventCaptor.getAllValues().get(0) instanceof DAG); assertEquals("dag_9999_0001_1", ((DAG)eventCaptor.getAllValues().get(0)).getID().toString()); DAGPlan dagPlan = DAGPlan.getDefaultInstance(); Map<String,LocalResource> localResources = new HashMap<String, LocalResource>(); dagClientHandler.submitDAG(dagPlan, localResources); verify(mockDagAM).submitDAGToAppMaster(dagPlan, localResources); dagClientHandler.shutdownAM(); verify(mockDagAM).shutdownTezAM(); }
public DAGClientHandler(DAGAppMaster dagAppMaster) { this.dagAppMaster = dagAppMaster; }
DAGClientHandler { public DAGClientHandler(DAGAppMaster dagAppMaster) { this.dagAppMaster = dagAppMaster; } }
DAGClientHandler { public DAGClientHandler(DAGAppMaster dagAppMaster) { this.dagAppMaster = dagAppMaster; } DAGClientHandler(DAGAppMaster dagAppMaster); }
DAGClientHandler { public DAGClientHandler(DAGAppMaster dagAppMaster) { this.dagAppMaster = dagAppMaster; } DAGClientHandler(DAGAppMaster dagAppMaster); List<String> getAllDAGs(); DAGStatus getDAGStatus(String dagIdStr,
      Set<StatusGetOpts> statusOptions); VertexStatus getVertexStatus(String dagIdStr, String vertexName,
      Set<StatusGetOpts> statusOptions); void tryKillDAG(String dagIdStr); synchronized String submitDAG(DAGPlan dagPlan,
      Map<String, LocalResource> additionalAmResources); synchronized void shutdownAM(); synchronized TezAppMasterStatus getSessionStatus(); synchronized void preWarmContainers(PreWarmContext preWarmContext); }
DAGClientHandler { public DAGClientHandler(DAGAppMaster dagAppMaster) { this.dagAppMaster = dagAppMaster; } DAGClientHandler(DAGAppMaster dagAppMaster); List<String> getAllDAGs(); DAGStatus getDAGStatus(String dagIdStr,
      Set<StatusGetOpts> statusOptions); VertexStatus getVertexStatus(String dagIdStr, String vertexName,
      Set<StatusGetOpts> statusOptions); void tryKillDAG(String dagIdStr); synchronized String submitDAG(DAGPlan dagPlan,
      Map<String, LocalResource> additionalAmResources); synchronized void shutdownAM(); synchronized TezAppMasterStatus getSessionStatus(); synchronized void preWarmContainers(PreWarmContext preWarmContext);  }