BackPressureStatsTracker { @SuppressWarnings("unchecked") public boolean triggerStackTraceSample(ExecutionJobVertex vertex) { synchronized (lock) { if (shutDown) { return false; } if (!pendingStats.contains(vertex) && !vertex.getGraph().getState().isTerminalState()) { ExecutionContext executionContext = vertex.getGraph().getExecutionContext(); if (executionContext != null) { pendingStats.add(vertex); if (LOG.isDebugEnabled()) { LOG.debug("Triggering stack trace sample for tasks: " + Arrays.toString(vertex.getTaskVertices())); } Future<StackTraceSample> sample = coordinator.triggerStackTraceSample( vertex.getTaskVertices(), numSamples, delayBetweenSamples, MAX_STACK_TRACE_DEPTH); sample.onComplete(new StackTraceSampleCompletionCallback(vertex), executionContext); return true; } } return false; } } BackPressureStatsTracker(
			StackTraceSampleCoordinator coordinator,
			int cleanUpInterval,
			int numSamples,
			FiniteDuration delayBetweenSamples); long getCleanUpInterval(); Option<OperatorBackPressureStats> getOperatorBackPressureStats(ExecutionJobVertex vertex); @SuppressWarnings("unchecked") boolean triggerStackTraceSample(ExecutionJobVertex vertex); void cleanUpOperatorStatsCache(); void shutDown();  }
@Test @SuppressWarnings("unchecked") public void testTriggerStackTraceSample() throws Exception { Promise<StackTraceSample> samplePromise = new Promise.DefaultPromise<>(); StackTraceSampleCoordinator sampleCoordinator = mock(StackTraceSampleCoordinator.class); when(sampleCoordinator.triggerStackTraceSample( any(ExecutionVertex[].class), anyInt(), any(FiniteDuration.class), anyInt())).thenReturn(samplePromise.future()); ExecutionGraph graph = mock(ExecutionGraph.class); when(graph.getState()).thenReturn(JobStatus.RUNNING); when(graph.getExecutionContext()).thenReturn(new ExecutionContext() { @Override public void execute(Runnable runnable) { runnable.run(); } @Override public void reportFailure(Throwable t) { fail(); } @Override public ExecutionContext prepare() { return this; } }); ExecutionVertex[] taskVertices = new ExecutionVertex[4]; ExecutionJobVertex jobVertex = mock(ExecutionJobVertex.class); when(jobVertex.getJobId()).thenReturn(new JobID()); when(jobVertex.getJobVertexId()).thenReturn(new JobVertexID()); when(jobVertex.getGraph()).thenReturn(graph); 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; FiniteDuration delayBetweenSamples = new FiniteDuration(100, TimeUnit.MILLISECONDS); BackPressureStatsTracker tracker = new BackPressureStatsTracker( sampleCoordinator, 9999, numSamples, delayBetweenSamples); assertTrue("Failed to trigger", tracker.triggerStackTraceSample(jobVertex)); verify(sampleCoordinator).triggerStackTraceSample( eq(taskVertices), eq(numSamples), eq(delayBetweenSamples), eq(BackPressureStatsTracker.MAX_STACK_TRACE_DEPTH)); assertFalse("Unexpected trigger", tracker.triggerStackTraceSample(jobVertex)); assertTrue(tracker.getOperatorBackPressureStats(jobVertex).isEmpty()); verify(sampleCoordinator).triggerStackTraceSample( eq(taskVertices), eq(numSamples), eq(delayBetweenSamples), eq(BackPressureStatsTracker.MAX_STACK_TRACE_DEPTH)); assertTrue(tracker.getOperatorBackPressureStats(jobVertex).isEmpty()); 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); samplePromise.success(sample); assertTrue(tracker.getOperatorBackPressureStats(jobVertex).isDefined()); OperatorBackPressureStats stats = tracker.getOperatorBackPressureStats(jobVertex).get(); assertEquals(sampleId, stats.getSampleId()); assertEquals(endTime, stats.getEndTimestamp()); assertEquals(taskVertices.length, stats.getNumberOfSubTasks()); for (int i = 0; i < taskVertices.length; i++) { double ratio = stats.getBackPressureRatio(i); assertEquals((i + 1) / ((double) 4), ratio, 0.0); } }