@Override
    protected CompletableFuture<JobSubmitResponseBody> handleRequest(
            @Nonnull HandlerRequest<JobSubmitRequestBody> request,
            @Nonnull DispatcherGateway gateway)
            throws RestHandlerException {
        final Collection<File> uploadedFiles = request.getUploadedFiles();
        final Map<String, Path> nameToFile =
                uploadedFiles.stream()
                        .collect(Collectors.toMap(File::getName, Path::fromLocalFile));

        if (uploadedFiles.size() != nameToFile.size()) {
            throw new RestHandlerException(
                    String.format(
                            "The number of uploaded files was %s than the expected count. Expected: %s Actual %s",
                            uploadedFiles.size() < nameToFile.size() ? "lower" : "higher",
                            nameToFile.size(),
                            uploadedFiles.size()),
                    HttpResponseStatus.BAD_REQUEST);
        }

        final JobSubmitRequestBody requestBody = request.getRequestBody();

        if (requestBody.jobGraphFileName == null) {
            throw new RestHandlerException(
                    String.format(
                            "The %s field must not be omitted or be null.",
                            JobSubmitRequestBody.FIELD_NAME_JOB_GRAPH),
                    HttpResponseStatus.BAD_REQUEST);
        }

        CompletableFuture<JobGraph> jobGraphFuture = loadJobGraph(requestBody, nameToFile);

        Collection<Path> jarFiles = getJarFilesToUpload(requestBody.jarFileNames, nameToFile);

        Collection<Tuple2<String, Path>> artifacts =
                getArtifactFilesToUpload(requestBody.artifactFileNames, nameToFile);

        CompletableFuture<JobGraph> finalizedJobGraphFuture =
                uploadJobGraphFiles(gateway, jobGraphFuture, jarFiles, artifacts, configuration);

        CompletableFuture<Acknowledge> jobSubmissionFuture =
                finalizedJobGraphFuture.thenCompose(
                        jobGraph -> gateway.submitJob(jobGraph, timeout));

        return jobSubmissionFuture.thenCombine(
                jobGraphFuture,
                (ack, jobGraph) -> new JobSubmitResponseBody("/jobs/" + jobGraph.getJobID()));
    }
@TestTemplate
    void testRejectionOnCountMismatch() throws Exception {
        final Path jobGraphFile = TempDirUtils.newFile(temporaryFolder).toPath();
        try (ObjectOutputStream objectOut =
                new ObjectOutputStream(Files.newOutputStream(jobGraphFile))) {
            objectOut.writeObject(JobGraphTestUtils.emptyJobGraph());
        }
        final Path countExceedingFile = TempDirUtils.newFile(temporaryFolder).toPath();

        TestingDispatcherGateway.Builder builder = TestingDispatcherGateway.newBuilder();
        builder.setBlobServerPort(blobServer.getPort())
                .setSubmitFunction(jobGraph -> CompletableFuture.completedFuture(Acknowledge.get()))
                .setHostname("localhost");
        DispatcherGateway mockGateway = builder.build();

        JobSubmitHandler handler =
                new JobSubmitHandler(
                        () -> CompletableFuture.completedFuture(mockGateway),
                        RpcUtils.INF_TIMEOUT,
                        Collections.emptyMap(),
                        Executors.directExecutor(),
                        configuration);

        JobSubmitRequestBody request =
                new JobSubmitRequestBody(
                        jobGraphFile.getFileName().toString(),
                        Collections.emptyList(),
                        Collections.emptyList());
        try {
            handler.handleRequest(
                            HandlerRequest.create(
                                    request,
                                    EmptyMessageParameters.getInstance(),
                                    Arrays.asList(
                                            jobGraphFile.toFile(), countExceedingFile.toFile())),
                            mockGateway)
                    .get();
        } catch (Exception e) {
            ExceptionUtils.findThrowable(
                    e,
                    candidate ->
                            candidate instanceof RestHandlerException
                                    && candidate.getMessage().contains("count"));
        }
    }