test_class.declarator: system
test_class.modifier: private static
test_class.original_string: private static ActorSystem system;
test_class.type: ActorSystem
test_class.var_name: system
test_class.declarator: coord
test_class.modifier: private
test_class.original_string: private StackTraceSampleCoordinator coord;
test_class.type: StackTraceSampleCoordinator
test_class.var_name: coord
test_class.file: flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/StackTraceSampleCoordinatorTest.java
test_class.identifier: StackTraceSampleCoordinatorTest
test_case.body: @Test
  	public void testTriggerStackTraceSample() throws Exception {
  		ExecutionVertex[] vertices = new ExecutionVertex[] {
  				mockExecutionVertex(new ExecutionAttemptID(), ExecutionState.RUNNING, true),
  				mockExecutionVertex(new ExecutionAttemptID(), ExecutionState.RUNNING, true),
  				mockExecutionVertex(new ExecutionAttemptID(), ExecutionState.RUNNING, true),
  				mockExecutionVertex(new ExecutionAttemptID(), ExecutionState.RUNNING, true)
  		};
  
  		int numSamples = 1;
  		FiniteDuration delayBetweenSamples = new FiniteDuration(100, TimeUnit.MILLISECONDS);
  		int maxStackTraceDepth = 0;
  
  		Future<StackTraceSample> sampleFuture = coord.triggerStackTraceSample(
  				vertices, numSamples, delayBetweenSamples, maxStackTraceDepth);
  
  		// Verify messages have been sent
  		for (ExecutionVertex vertex : vertices) {
  			ExecutionAttemptID expectedExecutionId = vertex
  					.getCurrentExecutionAttempt().getAttemptId();
  
  			TriggerStackTraceSample expectedMsg = new TriggerStackTraceSample(
  					0,
  					expectedExecutionId,
  					numSamples,
  					delayBetweenSamples,
  					maxStackTraceDepth);
  
  			verify(vertex).sendMessageToCurrentExecution(
  					eq(expectedMsg), eq(expectedExecutionId), any(AkkaActorGateway.class));
  		}
  
  		assertFalse(sampleFuture.isCompleted());
  
  		StackTraceElement[] stackTraceSample = Thread.currentThread().getStackTrace();
  		List<StackTraceElement[]> traces = new ArrayList<>();
  		traces.add(stackTraceSample);
  		traces.add(stackTraceSample);
  		traces.add(stackTraceSample);
  
  		// Collect stack traces
  		for (int i = 0; i < vertices.length; i++) {
  			ExecutionAttemptID executionId = vertices[i].getCurrentExecutionAttempt().getAttemptId();
  			coord.collectStackTraces(0, executionId, traces);
  
  			if (i == vertices.length - 1) {
  				assertTrue(sampleFuture.isCompleted());
  			} else {
  				assertFalse(sampleFuture.isCompleted());
  			}
  		}
  
  		// Verify completed stack trace sample
  		StackTraceSample sample = sampleFuture.value().get().get();
  
  		assertEquals(0, sample.getSampleId());
  		assertTrue(sample.getEndTime() >= sample.getStartTime());
  
  		Map<ExecutionAttemptID, List<StackTraceElement[]>> tracesByTask = sample.getStackTraces();
  
  		for (ExecutionVertex vertex : vertices) {
  			ExecutionAttemptID executionId = vertex.getCurrentExecutionAttempt().getAttemptId();
  			List<StackTraceElement[]> sampleTraces = tracesByTask.get(executionId);
  
  			assertNotNull("Task not found", sampleTraces);
  			assertTrue(traces.equals(sampleTraces));
  		}
  
  		// Verify no more pending sample
  		assertEquals(0, coord.getNumberOfPendingSamples());
  
  		// Verify no error on late collect
  		coord.collectStackTraces(0, vertices[0].getCurrentExecutionAttempt().getAttemptId(), traces);
  	}
test_case.class_method_signature: StackTraceSampleCoordinatorTest.testTriggerStackTraceSample()
test_case.full_signature: @Test public void testTriggerStackTraceSample()
test_case.identifier: testTriggerStackTraceSample
test_case.invocations: mockExecutionVertex mockExecutionVertex mockExecutionVertex mockExecutionVertex triggerStackTraceSample getAttemptId getCurrentExecutionAttempt sendMessageToCurrentExecution verify eq eq any assertFalse isCompleted getStackTrace currentThread add add add getAttemptId getCurrentExecutionAttempt collectStackTraces assertTrue isCompleted assertFalse isCompleted get get value assertEquals getSampleId assertTrue getEndTime getStartTime getStackTraces getAttemptId getCurrentExecutionAttempt get assertNotNull assertTrue equals assertEquals getNumberOfPendingSamples collectStackTraces getAttemptId getCurrentExecutionAttempt
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testTriggerStackTraceSample()
focal_class.declarator: LOG = LoggerFactory.getLogger(StackTraceSampleCoordinator.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Logger LOG = LoggerFactory.getLogger(StackTraceSampleCoordinator.class);
focal_class.type: Logger
focal_class.var_name: LOG
focal_class.declarator: NUM_GHOST_SAMPLE_IDS = 10
focal_class.modifier: private static final
focal_class.original_string: private static final int NUM_GHOST_SAMPLE_IDS = 10;
focal_class.type: int
focal_class.var_name: NUM_GHOST_SAMPLE_IDS
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: responseActor
focal_class.modifier: private final
focal_class.original_string: private final ActorGateway responseActor;
focal_class.type: ActorGateway
focal_class.var_name: responseActor
focal_class.declarator: sampleTimeout
focal_class.modifier: private final
focal_class.original_string: private final int sampleTimeout;
focal_class.type: int
focal_class.var_name: sampleTimeout
focal_class.declarator: pendingSamples = new HashMap<>()
focal_class.modifier: private final
focal_class.original_string: private final Map<Integer, PendingStackTraceSample> pendingSamples = new HashMap<>();
focal_class.type: Map<Integer, PendingStackTraceSample>
focal_class.var_name: pendingSamples
focal_class.declarator: recentPendingSamples = new ArrayDeque<>(NUM_GHOST_SAMPLE_IDS)
focal_class.modifier: private final
focal_class.original_string: private final ArrayDeque<Integer> recentPendingSamples = new ArrayDeque<>(NUM_GHOST_SAMPLE_IDS);
focal_class.type: ArrayDeque<Integer>
focal_class.var_name: recentPendingSamples
focal_class.declarator: sampleIdCounter
focal_class.modifier: private
focal_class.original_string: private int sampleIdCounter;
focal_class.type: int
focal_class.var_name: sampleIdCounter
focal_class.declarator: timer
focal_class.modifier: private
focal_class.original_string: private Timer timer;
focal_class.type: Timer
focal_class.var_name: timer
focal_class.declarator: isShutDown
focal_class.modifier: private
focal_class.original_string: private boolean isShutDown;
focal_class.type: boolean
focal_class.var_name: isShutDown
focal_class.file: flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/StackTraceSampleCoordinator.java
focal_class.identifier: StackTraceSampleCoordinator
focal_class.class_method_signature: StackTraceSampleCoordinator.StackTraceSampleCoordinator(ActorSystem actorSystem, int sampleTimeout)
focal_class.full_signature: public  StackTraceSampleCoordinator(ActorSystem actorSystem, int sampleTimeout)
focal_class.identifier: StackTraceSampleCoordinator
focal_class.modifiers: public
focal_class.parameters: (ActorSystem actorSystem, int sampleTimeout)
focal_class.signature:  StackTraceSampleCoordinator(ActorSystem actorSystem, int sampleTimeout)
focal_class.class_method_signature: StackTraceSampleCoordinator.triggerStackTraceSample(
  			ExecutionVertex[] tasksToSample,
  			int numSamples,
  			FiniteDuration delayBetweenSamples,
  			int maxStackTraceDepth)
focal_class.full_signature: @SuppressWarnings("unchecked") public Future<StackTraceSample> triggerStackTraceSample(
  			ExecutionVertex[] tasksToSample,
  			int numSamples,
  			FiniteDuration delayBetweenSamples,
  			int maxStackTraceDepth)
focal_class.identifier: triggerStackTraceSample
focal_class.modifiers: @SuppressWarnings("unchecked") public
focal_class.parameters: (
  			ExecutionVertex[] tasksToSample,
  			int numSamples,
  			FiniteDuration delayBetweenSamples,
  			int maxStackTraceDepth)
focal_class.return: Future<StackTraceSample>
focal_class.signature: Future<StackTraceSample> triggerStackTraceSample(
  			ExecutionVertex[] tasksToSample,
  			int numSamples,
  			FiniteDuration delayBetweenSamples,
  			int maxStackTraceDepth)
focal_class.class_method_signature: StackTraceSampleCoordinator.cancelStackTraceSample(int sampleId, Exception cause)
focal_class.full_signature: public void cancelStackTraceSample(int sampleId, Exception cause)
focal_class.identifier: cancelStackTraceSample
focal_class.modifiers: public
focal_class.parameters: (int sampleId, Exception cause)
focal_class.return: void
focal_class.signature: void cancelStackTraceSample(int sampleId, Exception cause)
focal_class.class_method_signature: StackTraceSampleCoordinator.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: StackTraceSampleCoordinator.collectStackTraces(
  			int sampleId,
  			ExecutionAttemptID executionId,
  			List<StackTraceElement[]> stackTraces)
focal_class.full_signature: public void collectStackTraces(
  			int sampleId,
  			ExecutionAttemptID executionId,
  			List<StackTraceElement[]> stackTraces)
focal_class.identifier: collectStackTraces
focal_class.modifiers: public
focal_class.parameters: (
  			int sampleId,
  			ExecutionAttemptID executionId,
  			List<StackTraceElement[]> stackTraces)
focal_class.return: void
focal_class.signature: void collectStackTraces(
  			int sampleId,
  			ExecutionAttemptID executionId,
  			List<StackTraceElement[]> stackTraces)
focal_class.class_method_signature: StackTraceSampleCoordinator.rememberRecentSampleId(int sampleId)
focal_class.full_signature: private void rememberRecentSampleId(int sampleId)
focal_class.identifier: rememberRecentSampleId
focal_class.modifiers: private
focal_class.parameters: (int sampleId)
focal_class.return: void
focal_class.signature: void rememberRecentSampleId(int sampleId)
focal_class.class_method_signature: StackTraceSampleCoordinator.getNumberOfPendingSamples()
focal_class.full_signature:  int getNumberOfPendingSamples()
focal_class.identifier: getNumberOfPendingSamples
focal_class.parameters: ()
focal_class.return: int
focal_class.signature: int getNumberOfPendingSamples()
focal_method.body: @SuppressWarnings("unchecked")
  	public Future<StackTraceSample> triggerStackTraceSample(
  			ExecutionVertex[] tasksToSample,
  			int numSamples,
  			FiniteDuration delayBetweenSamples,
  			int maxStackTraceDepth) {
  
  		checkNotNull(tasksToSample, "Tasks to sample");
  		checkArgument(tasksToSample.length >= 1, "No tasks to sample");
  		checkArgument(numSamples >= 1, "No number of samples");
  		checkArgument(maxStackTraceDepth >= 0, "Negative maximum stack trace depth");
  
  		// Execution IDs of running tasks
  		ExecutionAttemptID[] triggerIds = new ExecutionAttemptID[tasksToSample.length];
  
  		// Check that all tasks are RUNNING before triggering anything. The
  		// triggering can still fail.
  		for (int i = 0; i < triggerIds.length; i++) {
  			Execution execution = tasksToSample[i].getCurrentExecutionAttempt();
  			if (execution != null && execution.getState() == ExecutionState.RUNNING) {
  				triggerIds[i] = execution.getAttemptId();
  			} else {
  				Promise failedPromise = new scala.concurrent.impl.Promise.DefaultPromise<>()
  						.failure(new IllegalStateException("Task " + tasksToSample[i]
  								.getTaskNameWithSubtaskIndex() + " is not running."));
  				return failedPromise.future();
  			}
  		}
  
  		synchronized (lock) {
  			if (isShutDown) {
  				Promise failedPromise = new scala.concurrent.impl.Promise.DefaultPromise<>()
  						.failure(new IllegalStateException("Shut down"));
  				return failedPromise.future();
  			}
  
  			if (timer == null) {
  				timer = new Timer("Stack trace sample coordinator timer");
  			}
  
  			int sampleId = sampleIdCounter++;
  
  			LOG.debug("Triggering stack trace sample {}", sampleId);
  
  			final PendingStackTraceSample pending = new PendingStackTraceSample(
  					sampleId, triggerIds);
  
  			// Discard the sample if it takes too long. We don't send cancel
  			// messages to the task managers, but only wait for the responses
  			// and then ignore them.
  			long expectedDuration = numSamples * delayBetweenSamples.toMillis();
  			long discardDelay = expectedDuration + sampleTimeout;
  
  			TimerTask discardTask = new TimerTask() {
  				@Override
  				public void run() {
  					try {
  						synchronized (lock) {
  							if (!pending.isDiscarded()) {
  								LOG.info("Sample {} expired before completing",
  										pending.getSampleId());
  
  								pending.discard(new RuntimeException("Time out"));
  								if (pendingSamples.remove(pending.getSampleId()) != null) {
  									rememberRecentSampleId(pending.getSampleId());
  								}
  							}
  						}
  					} catch (Throwable t) {
  						LOG.error("Exception while handling sample timeout", t);
  					}
  				}
  			};
  
  			// Add the pending sample before scheduling the discard task to
  			// prevent races with removing it again.
  			pendingSamples.put(sampleId, pending);
  
  			timer.schedule(discardTask, discardDelay);
  
  			boolean success = true;
  			try {
  				// Trigger all samples
  				for (int i = 0; i < tasksToSample.length; i++) {
  					TriggerStackTraceSample msg = new TriggerStackTraceSample(
  							sampleId,
  							triggerIds[i],
  							numSamples,
  							delayBetweenSamples,
  							maxStackTraceDepth);
  
  					if (!tasksToSample[i].sendMessageToCurrentExecution(
  							msg,
  							triggerIds[i],
  							responseActor)) {
  						success = false;
  						break;
  					}
  				}
  
  				return pending.getStackTraceSampleFuture();
  			} finally {
  				if (!success) {
  					pending.discard(new RuntimeException("Failed to trigger sample, " +
  							"because task has been reset."));
  					pendingSamples.remove(sampleId);
  					rememberRecentSampleId(sampleId);
  				}
  			}
  		}
  	}
focal_method.class_method_signature: StackTraceSampleCoordinator.triggerStackTraceSample(
  			ExecutionVertex[] tasksToSample,
  			int numSamples,
  			FiniteDuration delayBetweenSamples,
  			int maxStackTraceDepth)
focal_method.full_signature: @SuppressWarnings("unchecked") public Future<StackTraceSample> triggerStackTraceSample(
  			ExecutionVertex[] tasksToSample,
  			int numSamples,
  			FiniteDuration delayBetweenSamples,
  			int maxStackTraceDepth)
focal_method.identifier: triggerStackTraceSample
focal_method.invocations: checkNotNull checkArgument checkArgument checkArgument getCurrentExecutionAttempt getState getAttemptId failure getTaskNameWithSubtaskIndex future failure future debug toMillis isDiscarded info getSampleId discard remove getSampleId rememberRecentSampleId getSampleId error put schedule sendMessageToCurrentExecution getStackTraceSampleFuture discard remove rememberRecentSampleId
focal_method.modifiers: @SuppressWarnings("unchecked") public
focal_method.parameters: (
  			ExecutionVertex[] tasksToSample,
  			int numSamples,
  			FiniteDuration delayBetweenSamples,
  			int maxStackTraceDepth)
focal_method.return: Future<StackTraceSample>
focal_method.signature: Future<StackTraceSample> triggerStackTraceSample(
  			ExecutionVertex[] tasksToSample,
  			int numSamples,
  			FiniteDuration delayBetweenSamples,
  			int maxStackTraceDepth)