StackTraceSampleCoordinator { public void shutDown() { synchronized (lock) { if (!isShutDown) { LOG.info("Shutting down stack trace sample coordinator."); for (PendingStackTraceSample pending : pendingSamples.values()) { pending.discard(new RuntimeException("Shut down")); } pendingSamples.clear(); if (timer != null) { timer.cancel(); } isShutDown = true; } } } StackTraceSampleCoordinator(ActorSystem actorSystem, int sampleTimeout); @SuppressWarnings("unchecked") Future<StackTraceSample> triggerStackTraceSample(
			ExecutionVertex[] tasksToSample,
			int numSamples,
			FiniteDuration delayBetweenSamples,
			int maxStackTraceDepth); void cancelStackTraceSample(int sampleId, Exception cause); void shutDown(); void collectStackTraces(
			int sampleId,
			ExecutionAttemptID executionId,
			List<StackTraceElement[]> stackTraces);  }
@Test public void testShutDown() throws Exception { ExecutionVertex[] vertices = new ExecutionVertex[] { mockExecutionVertex(new ExecutionAttemptID(), ExecutionState.RUNNING, true), }; List<Future<StackTraceSample>> sampleFutures = new ArrayList<>(); sampleFutures.add(coord.triggerStackTraceSample( vertices, 1, new FiniteDuration(100, TimeUnit.MILLISECONDS), 0)); sampleFutures.add(coord.triggerStackTraceSample( vertices, 1, new FiniteDuration(100, TimeUnit.MILLISECONDS), 0)); for (Future<StackTraceSample> future : sampleFutures) { assertFalse(future.isCompleted()); } coord.shutDown(); for (Future<StackTraceSample> future : sampleFutures) { assertTrue(future.isCompleted()); } Future<StackTraceSample> future = coord.triggerStackTraceSample( vertices, 1, new FiniteDuration(100, TimeUnit.MILLISECONDS), 0); assertTrue(future.isCompleted()); assertTrue(future.failed().isCompleted()); }