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 testShouldThrowExceptionIfTaskIsNotRunningBeforeSampling()
            throws ExecutionException, InterruptedException {
        Set<SampleableTask> tasks = new HashSet<>();
        tasks.add(new NotRunningTask());

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

        assertThatFuture(sampleFuture).eventuallyFails();
        assertThat(sampleFuture.handle((ignored, e) -> e).get())
                .isInstanceOf(IllegalStateException.class);
    }