test_class.file: flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/BackPressureStatsTrackerTest.java
test_class.identifier: BackPressureStatsTrackerTest
test_case.body: @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);
  
  		// Same Thread execution context
  		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);
  
  		// Trigger
  		assertTrue("Failed to trigger", tracker.triggerStackTraceSample(jobVertex));
  
  		verify(sampleCoordinator).triggerStackTraceSample(
  				eq(taskVertices),
  				eq(numSamples),
  				eq(delayBetweenSamples),
  				eq(BackPressureStatsTracker.MAX_STACK_TRACE_DEPTH));
  
  		// Trigger again for pending request, should not fire
  		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());
  
  		// Complete the future
  		Map<ExecutionAttemptID, List<StackTraceElement[]>> traces = new HashMap<>();
  		for (ExecutionVertex vertex : taskVertices) {
  			List<StackTraceElement[]> taskTraces = new ArrayList<>();
  
  			for (int i = 0; i < taskVertices.length; i++) {
  				// Traces until sub task index are back pressured
  				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);
  
  		// Succeed the promise
  		samplePromise.success(sample);
  
  		assertTrue(tracker.getOperatorBackPressureStats(jobVertex).isDefined());
  
  		OperatorBackPressureStats stats = tracker.getOperatorBackPressureStats(jobVertex).get();
  
  		// Verify the stats
  		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);
  			// Traces until sub task index are back pressured
  			assertEquals((i + 1) / ((double) 4), ratio, 0.0);
  		}
  	}
test_case.class_method_signature: BackPressureStatsTrackerTest.testTriggerStackTraceSample()
test_case.full_signature: @Test @SuppressWarnings("unchecked") public void testTriggerStackTraceSample()
test_case.identifier: testTriggerStackTraceSample
test_case.invocations: mock thenReturn when triggerStackTraceSample any anyInt any anyInt future mock thenReturn when getState thenReturn when getExecutionContext run fail mock thenReturn when getJobId thenReturn when getJobVertexId thenReturn when getGraph thenReturn when getTaskVertices mockExecutionVertex mockExecutionVertex mockExecutionVertex mockExecutionVertex assertTrue triggerStackTraceSample triggerStackTraceSample verify eq eq eq eq assertFalse triggerStackTraceSample assertTrue isEmpty getOperatorBackPressureStats triggerStackTraceSample verify eq eq eq eq assertTrue isEmpty getOperatorBackPressureStats add createStackTrace getParallelSubtaskIndex put getAttemptId getCurrentExecutionAttempt success assertTrue isDefined getOperatorBackPressureStats get getOperatorBackPressureStats assertEquals getSampleId assertEquals getEndTimestamp assertEquals getNumberOfSubTasks getBackPressureRatio assertEquals
test_case.modifiers: @Test @SuppressWarnings("unchecked") public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testTriggerStackTraceSample()
focal_class.declarator: LOG = LoggerFactory.getLogger(BackPressureStatsTracker.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Logger LOG = LoggerFactory.getLogger(BackPressureStatsTracker.class);
focal_class.type: Logger
focal_class.var_name: LOG
focal_class.declarator: MAX_STACK_TRACE_DEPTH = 3
focal_class.modifier: static final
focal_class.original_string: static final int MAX_STACK_TRACE_DEPTH = 3;
focal_class.type: int
focal_class.var_name: MAX_STACK_TRACE_DEPTH
focal_class.declarator: EXPECTED_CLASS_NAME = "org.apache.flink.runtime.io.network.buffer.LocalBufferPool"
focal_class.modifier: static final
focal_class.original_string: static final String EXPECTED_CLASS_NAME = "org.apache.flink.runtime.io.network.buffer.LocalBufferPool";
focal_class.type: String
focal_class.var_name: EXPECTED_CLASS_NAME
focal_class.declarator: EXPECTED_METHOD_NAME = "requestBufferBlocking"
focal_class.modifier: static final
focal_class.original_string: static final String EXPECTED_METHOD_NAME = "requestBufferBlocking";
focal_class.type: String
focal_class.var_name: EXPECTED_METHOD_NAME
focal_class.declarator: lock = new Object()
focal_class.modifier: private final
focal_class.original_string: private final Object lock = new Object();
focal_class.type: Object
focal_class.var_name: lock
focal_class.declarator: coordinator
focal_class.modifier: private final
focal_class.original_string: private final StackTraceSampleCoordinator coordinator;
focal_class.type: StackTraceSampleCoordinator
focal_class.var_name: coordinator
focal_class.declarator: operatorStatsCache
focal_class.modifier: private final
focal_class.original_string: private final Cache<ExecutionJobVertex, OperatorBackPressureStats> operatorStatsCache;
focal_class.type: Cache<ExecutionJobVertex, OperatorBackPressureStats>
focal_class.var_name: operatorStatsCache
focal_class.declarator: pendingStats = new HashSet<>()
focal_class.modifier: private final
focal_class.original_string: private final Set<ExecutionJobVertex> pendingStats = new HashSet<>();
focal_class.type: Set<ExecutionJobVertex>
focal_class.var_name: pendingStats
focal_class.declarator: cleanUpInterval
focal_class.modifier: private final
focal_class.original_string: private final int cleanUpInterval;
focal_class.type: int
focal_class.var_name: cleanUpInterval
focal_class.declarator: numSamples
focal_class.modifier: private final
focal_class.original_string: private final int numSamples;
focal_class.type: int
focal_class.var_name: numSamples
focal_class.declarator: delayBetweenSamples
focal_class.modifier: private final
focal_class.original_string: private final FiniteDuration delayBetweenSamples;
focal_class.type: FiniteDuration
focal_class.var_name: delayBetweenSamples
focal_class.declarator: shutDown
focal_class.modifier: private
focal_class.original_string: private boolean shutDown;
focal_class.type: boolean
focal_class.var_name: shutDown
focal_class.file: flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/BackPressureStatsTracker.java
focal_class.identifier: BackPressureStatsTracker
focal_class.class_method_signature: BackPressureStatsTracker.BackPressureStatsTracker(
  			StackTraceSampleCoordinator coordinator,
  			int cleanUpInterval,
  			int numSamples,
  			FiniteDuration delayBetweenSamples)
focal_class.full_signature: public  BackPressureStatsTracker(
  			StackTraceSampleCoordinator coordinator,
  			int cleanUpInterval,
  			int numSamples,
  			FiniteDuration delayBetweenSamples)
focal_class.identifier: BackPressureStatsTracker
focal_class.modifiers: public
focal_class.parameters: (
  			StackTraceSampleCoordinator coordinator,
  			int cleanUpInterval,
  			int numSamples,
  			FiniteDuration delayBetweenSamples)
focal_class.signature:  BackPressureStatsTracker(
  			StackTraceSampleCoordinator coordinator,
  			int cleanUpInterval,
  			int numSamples,
  			FiniteDuration delayBetweenSamples)
focal_class.class_method_signature: BackPressureStatsTracker.getCleanUpInterval()
focal_class.full_signature: public long getCleanUpInterval()
focal_class.identifier: getCleanUpInterval
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: long
focal_class.signature: long getCleanUpInterval()
focal_class.class_method_signature: BackPressureStatsTracker.getOperatorBackPressureStats(ExecutionJobVertex vertex)
focal_class.full_signature: public Option<OperatorBackPressureStats> getOperatorBackPressureStats(ExecutionJobVertex vertex)
focal_class.identifier: getOperatorBackPressureStats
focal_class.modifiers: public
focal_class.parameters: (ExecutionJobVertex vertex)
focal_class.return: Option<OperatorBackPressureStats>
focal_class.signature: Option<OperatorBackPressureStats> getOperatorBackPressureStats(ExecutionJobVertex vertex)
focal_class.class_method_signature: BackPressureStatsTracker.triggerStackTraceSample(ExecutionJobVertex vertex)
focal_class.full_signature: @SuppressWarnings("unchecked") public boolean triggerStackTraceSample(ExecutionJobVertex vertex)
focal_class.identifier: triggerStackTraceSample
focal_class.modifiers: @SuppressWarnings("unchecked") public
focal_class.parameters: (ExecutionJobVertex vertex)
focal_class.return: boolean
focal_class.signature: boolean triggerStackTraceSample(ExecutionJobVertex vertex)
focal_class.class_method_signature: BackPressureStatsTracker.cleanUpOperatorStatsCache()
focal_class.full_signature: public void cleanUpOperatorStatsCache()
focal_class.identifier: cleanUpOperatorStatsCache
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void cleanUpOperatorStatsCache()
focal_class.class_method_signature: BackPressureStatsTracker.shutDown()
focal_class.full_signature: public void shutDown()
focal_class.identifier: shutDown
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void shutDown()
focal_class.class_method_signature: BackPressureStatsTracker.invalidateOperatorStatsCache()
focal_class.full_signature:  void invalidateOperatorStatsCache()
focal_class.identifier: invalidateOperatorStatsCache
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void invalidateOperatorStatsCache()
focal_method.body: @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();
  
  				// Only trigger if still active job
  				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;
  		}
  	}
focal_method.class_method_signature: BackPressureStatsTracker.triggerStackTraceSample(ExecutionJobVertex vertex)
focal_method.full_signature: @SuppressWarnings("unchecked") public boolean triggerStackTraceSample(ExecutionJobVertex vertex)
focal_method.identifier: triggerStackTraceSample
focal_method.invocations: contains isTerminalState getState getGraph getExecutionContext getGraph add isDebugEnabled debug toString getTaskVertices triggerStackTraceSample getTaskVertices onComplete
focal_method.modifiers: @SuppressWarnings("unchecked") public
focal_method.parameters: (ExecutionJobVertex vertex)
focal_method.return: boolean
focal_method.signature: boolean triggerStackTraceSample(ExecutionJobVertex vertex)