You are certainly right, that this is not the optimal place. In fact, most of that was already present and I only added a guard for distributed FS at this place. I was looking for a quick fix that doesn't break anything shortly before the release and didn't want to change too many things.

Now with the release out of the way, I can look for a proper solution based on file systems. However, the distinction between distributed and local file system will most likely remain in the DataSinkTask and the FileInputFormat as the distributed FS should be initialized at planning time (within the JM) and local FS need to be initialized in parallel at runtime...