public CompletableFuture<VertexThreadInfoStats> triggerThreadInfoRequest(
            Map<ImmutableSet<ExecutionAttemptID>, CompletableFuture<TaskExecutorThreadInfoGateway>>
                    executionsWithGateways,
            int numSamples,
            Duration delayBetweenSamples,
            int maxStackTraceDepth) {

        checkNotNull(executionsWithGateways, "Tasks to sample");
        checkArgument(executionsWithGateways.size() > 0, "No tasks to sample");
        checkArgument(numSamples >= 1, "No number of samples");
        checkArgument(maxStackTraceDepth >= 0, "Negative maximum stack trace depth");

        // Execution IDs of running tasks grouped by the task manager
        Collection<ImmutableSet<ExecutionAttemptID>> runningSubtasksIds =
                executionsWithGateways.keySet();

        synchronized (lock) {
            if (isShutDown) {
                return FutureUtils.completedExceptionally(new IllegalStateException("Shut down"));
            }

            final int requestId = requestIdCounter++;

            log.debug("Triggering thread info request {}", requestId);

            final PendingThreadInfoRequest pending =
                    new PendingThreadInfoRequest(requestId, runningSubtasksIds);

            // requestTimeout is treated as the time on top of the expected sampling duration.
            // Discard the request 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();
            Time timeout = Time.milliseconds(expectedDuration + requestTimeout.toMillis());

            // Add the pending request before scheduling the discard task to
            // prevent races with removing it again.
            pendingRequests.put(requestId, pending);

            ThreadInfoSamplesRequest requestParams =
                    new ThreadInfoSamplesRequest(
                            requestId, numSamples, delayBetweenSamples, maxStackTraceDepth);

            requestThreadInfo(executionsWithGateways, requestParams, timeout);

            return pending.getStatsFuture();
        }
    }
@Test
    void testThreadInfoRequestWithException() throws Exception {
        Map<ImmutableSet<ExecutionAttemptID>, CompletableFuture<TaskExecutorThreadInfoGateway>>
                executionWithGateways =
                        createMockSubtaskWithGateways(
                                CompletionType.SUCCESSFULLY, CompletionType.EXCEPTIONALLY);

        CompletableFuture<VertexThreadInfoStats> requestFuture =
                coordinator.triggerThreadInfoRequest(
                        executionWithGateways,
                        DEFAULT_NUMBER_OF_SAMPLES,
                        DEFAULT_DELAY_BETWEEN_SAMPLES,
                        DEFAULT_MAX_STACK_TRACE_DEPTH);

        assertThatThrownBy(requestFuture::get, "The request must be failed.")
                .isInstanceOf(ExecutionException.class)
                .hasCauseInstanceOf(RuntimeException.class);
    }