public CompletableFuture<Map<ExecutionAttemptID, Collection<ThreadInfoSample>>>
            requestThreadInfoSamples(
                    Map<Long, ExecutionAttemptID> threads,
                    final ThreadInfoSamplesRequest requestParams) {
        checkNotNull(threads, "threads must not be null");
        checkNotNull(requestParams, "requestParams must not be null");

        CompletableFuture<Map<ExecutionAttemptID, Collection<ThreadInfoSample>>> resultFuture =
                new CompletableFuture<>();
        scheduledExecutor.execute(
                () ->
                        requestThreadInfoSamples(
                                threads,
                                requestParams.getNumSamples(),
                                requestParams.getDelayBetweenSamples(),
                                requestParams.getMaxStackTraceDepth(),
                                CollectionUtil.newHashMapWithExpectedSize(threads.size()),
                                resultFuture));
        return resultFuture;
    }
@Test
    void testSampleTaskThreadInfo() throws Exception {
        Set<IdleTestTask> tasks = new HashSet<>();
        executeWithTerminationGuarantee(
                () -> {
                    tasks.add(new IdleTestTask());
                    tasks.add(new IdleTestTask());
                    Thread.sleep(2000);

                    Map<Long, ExecutionAttemptID> threads = collectExecutionAttempts(tasks);
                    final Map<ExecutionAttemptID, Collection<ThreadInfoSample>> threadInfoSamples =
                            threadInfoSampleService
                                    .requestThreadInfoSamples(threads, requestParams)
                                    .get();

                    int count = 0;
                    for (Collection<ThreadInfoSample> samples : threadInfoSamples.values()) {
                        for (ThreadInfoSample sample : samples) {
                            count++;
                            StackTraceElement[] traces = sample.getStackTrace();
                            assertThat(traces).hasSizeLessThanOrEqualTo(MAX_STACK_TRACK_DEPTH);
                        }
                    }
                    assertThat(count).isEqualTo(NUMBER_OF_SAMPLES * 2);
                },
                tasks);
    }