@Override
    public CompletableFuture<TransientBlobKey> requestFileUploadByType(
            FileType fileType, Time timeout) {
        final String filePath;
        switch (fileType) {
            case LOG:
                filePath = taskManagerConfiguration.getTaskManagerLogPath();
                break;
            case STDOUT:
                filePath = taskManagerConfiguration.getTaskManagerStdoutPath();
                break;
            default:
                filePath = null;
        }
        return requestFileUploadByFilePath(filePath, fileType.toString());
    }
@Test
    @Timeout(10)
    void testLogNotFoundHandling() throws Throwable {
        configuration.set(NettyShuffleEnvironmentOptions.DATA_PORT, 0);
        configuration.set(NettyShuffleEnvironmentOptions.NETWORK_REQUEST_BACKOFF_INITIAL, 100);
        configuration.set(NettyShuffleEnvironmentOptions.NETWORK_REQUEST_BACKOFF_MAX, 200);
        configuration.set(TASK_MANAGER_LOG_PATH, "/i/dont/exist");

        try (TaskSubmissionTestEnvironment env =
                new Builder(jobId)
                        .setConfiguration(configuration)
                        .setLocalCommunication(false)
                        .build(EXECUTOR_EXTENSION.getExecutor())) {
            TaskExecutorGateway tmGateway = env.getTaskExecutorGateway();
            CompletableFuture<TransientBlobKey> logFuture =
                    tmGateway.requestFileUploadByType(FileType.LOG, timeout);

            assertThatFuture(logFuture)
                    .eventuallyFails()
                    .withThrowableOfType(ExecutionException.class)
                    .withMessageContaining("The file LOG does not exist on the TaskExecutor.");
        }
    }