@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;
	Time delayBetweenSamples = Time.milliseconds(100L);
	int maxStackTraceDepth = 0;
	CompletableFuture<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);
		Mockito.verify(vertex.getCurrentExecutionAttempt())
			.requestStackTraceSample(Matchers.eq(0), Matchers.eq(numSamples), Matchers.eq(delayBetweenSamples), Matchers.eq(maxStackTraceDepth), Matchers.any(Time.class));
	}
	Assert.assertFalse(sampleFuture.isDone());
	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) {
			Assert.assertTrue(sampleFuture.isDone());
		} else {
			Assert.assertFalse(sampleFuture.isDone());
		}
	}
	// Verify completed stack trace sample
	StackTraceSample sample = sampleFuture.get();
	Assert.assertEquals(0, sample.getSampleId());
	Assert.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);
		Assert.assertNotNull("Task not found", sampleTraces);
		Assert.assertTrue(traces.equals(sampleTraces));
	}
	// Verify no more pending sample
	Assert.assertEquals(0, coord.getNumberOfPendingSamples());
	// Verify no error on late collect
	coord.collectStackTraces(0, vertices[0].getCurrentExecutionAttempt().getAttemptId(), traces);
}