flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/backpressure/BackPressureStatsTrackerTest.java
/** Tests simple statistics with fake stack traces. */
Tests simple statistics with fake stack traces.
Tests simple statistics with fake stack traces .
@Test
	@SuppressWarnings("unchecked")
	public void testTriggerStackTraceSample() throws Exception {
		CompletableFuture<StackTraceSample> sampleFuture = new CompletableFuture<>();
		StackTraceSampleCoordinator sampleCoordinator = Mockito.mock(StackTraceSampleCoordinator.class);
		Mockito.when(sampleCoordinator.triggerStackTraceSample(
				Matchers.any(ExecutionVertex[].class),
				Matchers.anyInt(),
				Matchers.any(Time.class),
				Matchers.anyInt())).thenReturn(sampleFuture);
		ExecutionGraph graph = Mockito.mock(ExecutionGraph.class);
		Mockito.when(graph.getState()).thenReturn(JobStatus.RUNNING);
		Mockito.when(graph.getFutureExecutor()).thenReturn(new Executor() {
			@Override
			public void execute(Runnable runnable) {
				runnable.run();
			}
		});
		ExecutionVertex[] taskVertices = new ExecutionVertex[4];
		ExecutionJobVertex jobVertex = Mockito.mock(ExecutionJobVertex.class);
		Mockito.when(jobVertex.getJobId()).thenReturn(new JobID());
		Mockito.when(jobVertex.getJobVertexId()).thenReturn(new JobVertexID());
		Mockito.when(jobVertex.getGraph()).thenReturn(graph);
		Mockito.when(jobVertex.getTaskVertices()).thenReturn(taskVertices);
		taskVertices[0] = mockExecutionVertex(jobVertex, 0);
		taskVertices[1] = mockExecutionVertex(jobVertex, 1);
		taskVertices[2] = mockExecutionVertex(jobVertex, 2);
		taskVertices[3] = mockExecutionVertex(jobVertex, 3);
		int numSamples = 100;
		Time delayBetweenSamples = Time.milliseconds(100L);
		BackPressureStatsTracker tracker = new BackPressureStatsTracker(
				sampleCoordinator, 9999, numSamples, delayBetweenSamples);
		Assert.assertTrue("Failed to trigger", tracker.triggerStackTraceSample(jobVertex));
		Mockito.verify(sampleCoordinator).triggerStackTraceSample(
				Matchers.eq(taskVertices),
				Matchers.eq(numSamples),
				Matchers.eq(delayBetweenSamples),
				Matchers.eq(BackPressureStatsTracker.MAX_STACK_TRACE_DEPTH));
		 again for pending request, should not fire
		Assert.assertFalse("Unexpected trigger", tracker.triggerStackTraceSample(jobVertex));
		Assert.assertTrue(!tracker.getOperatorBackPressureStats(jobVertex).isPresent());
		Mockito.verify(sampleCoordinator).triggerStackTraceSample(
				Matchers.eq(taskVertices),
				Matchers.eq(numSamples),
				Matchers.eq(delayBetweenSamples),
				Matchers.eq(BackPressureStatsTracker.MAX_STACK_TRACE_DEPTH));
		Assert.assertTrue(!tracker.getOperatorBackPressureStats(jobVertex).isPresent());
		Map<ExecutionAttemptID, List<StackTraceElement[]>> traces = new HashMap<>();
		for (ExecutionVertex vertex : taskVertices) {
			List<StackTraceElement[]> taskTraces = new ArrayList<>();
			for (int i = 0; i < taskVertices.length; i++) {
				taskTraces.add(createStackTrace(i <= vertex.getParallelSubtaskIndex()));
			}
			traces.put(vertex.getCurrentExecutionAttempt().getAttemptId(), taskTraces);
		}
		int sampleId = 1231;
		int endTime = 841;
		StackTraceSample sample = new StackTraceSample(
				sampleId,
				0,
				endTime,
				traces);
		sampleFuture.complete(sample);
		Assert.assertTrue(tracker.getOperatorBackPressureStats(jobVertex).isPresent());
		OperatorBackPressureStats stats = tracker.getOperatorBackPressureStats(jobVertex).get();
		Assert.assertEquals(sampleId, stats.getSampleId());
		Assert.assertEquals(endTime, stats.getEndTimestamp());
		Assert.assertEquals(taskVertices.length, stats.getNumberOfSubTasks());
		for (int i = 0; i < taskVertices.length; i++) {
			double ratio = stats.getBackPressureRatio(i);
			Assert.assertEquals((i + 1) / ((double) 4), ratio, 0.0);
		}
	}
@ Test @ SuppressWarnings ( "unchecked" ) public void testTriggerStackTraceSample ( ) throws Exception { CompletableFuture < StackTraceSample > sampleFuture = new CompletableFuture < > ( ) ; StackTraceSampleCoordinator sampleCoordinator = Mockito . mock ( StackTraceSampleCoordinator . class ) ; Mockito . when ( sampleCoordinator . triggerStackTraceSample ( Matchers . any ( ExecutionVertex [ ] . class ) , Matchers . anyInt ( ) , Matchers . any ( Time . class ) , Matchers . anyInt ( ) ) ) . thenReturn ( sampleFuture ) ; ExecutionGraph graph = Mockito . mock ( ExecutionGraph . class ) ; Mockito . when ( graph . getState ( ) ) . thenReturn ( JobStatus . RUNNING ) ; Mockito . when ( graph . getFutureExecutor ( ) ) . thenReturn ( new Executor ( ) { @ Override public void execute ( Runnable runnable ) { runnable . run ( ) ; } } ) ; ExecutionVertex [ ] taskVertices = new ExecutionVertex [ 4 ] ; ExecutionJobVertex jobVertex = Mockito . mock ( ExecutionJobVertex . class ) ; Mockito . when ( jobVertex . getJobId ( ) ) . thenReturn ( new JobID ( ) ) ; Mockito . when ( jobVertex . getJobVertexId ( ) ) . thenReturn ( new JobVertexID ( ) ) ; Mockito . when ( jobVertex . getGraph ( ) ) . thenReturn ( graph ) ; Mockito . when ( jobVertex . getTaskVertices ( ) ) . thenReturn ( taskVertices ) ; taskVertices [ 0 ] = mockExecutionVertex ( jobVertex , 0 ) ; taskVertices [ 1 ] = mockExecutionVertex ( jobVertex , 1 ) ; taskVertices [ 2 ] = mockExecutionVertex ( jobVertex , 2 ) ; taskVertices [ 3 ] = mockExecutionVertex ( jobVertex , 3 ) ; int numSamples = 100 ; Time delayBetweenSamples = Time . milliseconds ( 100L ) ; BackPressureStatsTracker tracker = new BackPressureStatsTracker ( sampleCoordinator , 9999 , numSamples , delayBetweenSamples ) ; Assert . assertTrue ( "Failed to trigger" , tracker . triggerStackTraceSample ( jobVertex ) ) ; Mockito . verify ( sampleCoordinator ) . triggerStackTraceSample ( Matchers . eq ( taskVertices ) , Matchers . eq ( numSamples ) , Matchers . eq ( delayBetweenSamples ) , Matchers . eq ( BackPressureStatsTracker . MAX_STACK_TRACE_DEPTH ) ) ; Assert . assertFalse ( "Unexpected trigger" , tracker . triggerStackTraceSample ( jobVertex ) ) ; Assert . assertTrue ( ! tracker . getOperatorBackPressureStats ( jobVertex ) . isPresent ( ) ) ; Mockito . verify ( sampleCoordinator ) . triggerStackTraceSample ( Matchers . eq ( taskVertices ) , Matchers . eq ( numSamples ) , Matchers . eq ( delayBetweenSamples ) , Matchers . eq ( BackPressureStatsTracker . MAX_STACK_TRACE_DEPTH ) ) ; Assert . assertTrue ( ! tracker . getOperatorBackPressureStats ( jobVertex ) . isPresent ( ) ) ; Map < ExecutionAttemptID , List < StackTraceElement [ ] > > traces = new HashMap < > ( ) ; for ( ExecutionVertex vertex : taskVertices ) { List < StackTraceElement [ ] > taskTraces = new ArrayList < > ( ) ; for ( int i = 0 ; i < taskVertices . length ; i ++ ) { taskTraces . add ( createStackTrace ( i <= vertex . getParallelSubtaskIndex ( ) ) ) ; } traces . put ( vertex . getCurrentExecutionAttempt ( ) . getAttemptId ( ) , taskTraces ) ; } int sampleId = 1231 ; int endTime = 841 ; StackTraceSample sample = new StackTraceSample ( sampleId , 0 , endTime , traces ) ; sampleFuture . complete ( sample ) ; Assert . assertTrue ( tracker . getOperatorBackPressureStats ( jobVertex ) . isPresent ( ) ) ; OperatorBackPressureStats stats = tracker . getOperatorBackPressureStats ( jobVertex ) . get ( ) ; Assert . assertEquals ( sampleId , stats . getSampleId ( ) ) ; Assert . assertEquals ( endTime , stats . getEndTimestamp ( ) ) ; Assert . assertEquals ( taskVertices . length , stats . getNumberOfSubTasks ( ) ) ; for ( int i = 0 ; i < taskVertices . length ; i ++ ) { double ratio = stats . getBackPressureRatio ( i ) ; Assert . assertEquals ( ( i + 1 ) / ( ( double ) 4 ) , ratio , 0.0 ) ; } }
Tests simple statistics with fake stack traces.
Tests simple statistics with fake stack traces .
// Same Thread execution context // Trigger // Trigger again for pending request, should not fire // Complete the future // Traces until sub task index are back pressured // Succeed the promise // Verify the stats // Traces until sub task index are back pressured