docs: @SuppressWarnings("unchecked")
  	public CompletableFuture<StackTraceSample> triggerStackTraceSample(
  			ExecutionVertex[] tasksToSample,
  			int numSamples,
  			Time 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];
  		Execution[] executions = new Execution[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) {
  				executions[i] = execution;
  				triggerIds[i] = execution.getAttemptId();
  			} else {
  				return FutureUtils.completedExceptionally(new IllegalStateException("Task " + tasksToSample[i]
  					.getTaskNameWithSubtaskIndex() + " is not running."));
  			}
  		}
  
  		synchronized (lock) {
  			if (isShutDown) {
  				return FutureUtils.completedExceptionally(new IllegalStateException("Shut down"));
  			}
  
  			final 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.toMilliseconds();
  			Time timeout = Time.milliseconds(expectedDuration + sampleTimeout);
  
  			// Add the pending sample before scheduling the discard task to
  			// prevent races with removing it again.
  			pendingSamples.put(sampleId, pending);
  
  			// Trigger all samples
  			for (Execution execution: executions) {
  				final CompletableFuture<StackTraceSampleResponse> stackTraceSampleFuture = execution.requestStackTraceSample(
  					sampleId,
  					numSamples,
  					delayBetweenSamples,
  					maxStackTraceDepth,
  					timeout);
  
  				stackTraceSampleFuture.handleAsync(
  					(StackTraceSampleResponse stackTraceSampleResponse, Throwable throwable) -> {
  						if (stackTraceSampleResponse != null) {
  							collectStackTraces(
  								stackTraceSampleResponse.getSampleId(),
  								stackTraceSampleResponse.getExecutionAttemptID(),
  								stackTraceSampleResponse.getSamples());
  						} else {
  							cancelStackTraceSample(sampleId, throwable);
  						}
  
  						return null;
  					},
  					executor);
  			}
  
  			return pending.getStackTraceSampleFuture();
  		}
  	}