@Override
    protected CompletableFuture<ProfilingInfo> handleRequest(
            @Nonnull HandlerRequest<ProfilingRequestBody> request,
            @Nonnull ResourceManagerGateway gateway)
            throws RestHandlerException {
        ProfilingRequestBody profilingRequest = request.getRequestBody();
        int duration = profilingRequest.getDuration();
        if (duration <= 0 || duration > maxDurationInSeconds) {
            return FutureUtils.completedExceptionally(
                    new IllegalArgumentException(
                            String.format(
                                    "`duration` must be set between (0s, %ds].",
                                    maxDurationInSeconds)));
        }
        final ResourceID taskManagerId = request.getPathParameter(TaskManagerIdPathParameter.class);
        return gateway.requestProfiling(
                taskManagerId, duration, profilingRequest.getMode(), getTimeout());
    }
@Test
    void testGetTaskManagerProfilingForUnknownTaskExecutorException() throws Exception {
        resourceManagerGateway.setRequestProfilingListFunction(
                EXPECTED_TASK_MANAGER_ID ->
                        FutureUtils.completedExceptionally(
                                new UnknownTaskExecutorException(EXPECTED_TASK_MANAGER_ID)));
        try {
            taskManagerProfilingHandler.handleRequest(handlerRequest, resourceManagerGateway).get();
        } catch (ExecutionException e) {
            final Throwable cause = e.getCause();
            assertThat(cause).isInstanceOf(UnknownTaskExecutorException.class);

            final UnknownTaskExecutorException unknownTaskExecutorException =
                    (UnknownTaskExecutorException) cause;
            assertThat(unknownTaskExecutorException.getMessage())
                    .contains("No TaskExecutor registered under " + EXPECTED_TASK_MANAGER_ID);
        }
    }