[FLINK-14908] Distributing CacheFiles through DFS does not work

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Duplicate
• Affects Version/s: 1.8.2, 1.9.1
• Component/s: Client / Job Submission, Runtime / REST
• Assignee: Unassigned
• Reporter: Dawid Wysakowicz
• Created: 21/Nov/19 16:41
• Updated: 22/Nov/19 12:16
• Resolved: 22/Nov/19 12:16

Description

User reported that distributing cache files through DFS does not work anymore: https://stackoverflow.com/questions/58978476/flink-1-9-wont-run-program-when-i-use-distributed-cache-why

I think the problematic part is in RestClusterClient#submitJob:

for (Map.Entry<String, DistributedCache.DistributedCacheEntry> artifacts : jobGraph.getUserArtifacts().entrySet()) {
	artifactFileNames.add(new JobSubmitRequestBody.DistributedCacheFile(artifacts.getKey(), new Path(artifacts.getValue().filePath).getName()));
	filesToUpload.add(new FileUpload(Paths.get(artifacts.getValue().filePath), RestConstants.CONTENT_TYPE_BINARY));
}

The code does not check if a file is in DFS, but just assumes it is in local FS and tries to add it to the rest request, which fails. The code on the receiver side in JobSubmitHandler still can support files distributed via DFS, but need to get proper paths to files in DFS.

Issue Links

• duplicates: FLINK-13758 Failed to submit JobGraph when registered hdfs file in DistributedCache