BackPressureStatsTracker { @SuppressWarnings("unchecked") public boolean triggerStackTraceSample(ExecutionJobVertex vertex) { synchronized (lock) { if (shutDown) { return false; } if (!pendingStats.contains(vertex) && !vertex.getGraph().getState().isGloballyTerminalState()) { Executor executor = vertex.getGraph().getFutureExecutor(); if (executor != null) { pendingStats.add(vertex); if (LOG.isDebugEnabled()) { LOG.debug("Triggering stack trace sample for tasks: " + Arrays.toString(vertex.getTaskVertices())); } CompletableFuture<StackTraceSample> sample = coordinator.triggerStackTraceSample( vertex.getTaskVertices(), numSamples, delayBetweenSamples, MAX_STACK_TRACE_DEPTH); sample.handleAsync(new StackTraceSampleCompletionCallback(vertex), executor); return true; } } return false; } } BackPressureStatsTracker(
			StackTraceSampleCoordinator coordinator,
			int cleanUpInterval,
			int numSamples,
			Time delayBetweenSamples); long getCleanUpInterval(); Optional<OperatorBackPressureStats> getOperatorBackPressureStats(ExecutionJobVertex vertex); @SuppressWarnings("unchecked") boolean triggerStackTraceSample(ExecutionJobVertex vertex); void cleanUpOperatorStatsCache(); void shutDown();  }
@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); } }