BucketingSink does not detect hadoop file systems

In BucketingSink#createHadoopFileSystem one can find this piece of code:

final org.apache.flink.core.fs.FileSystem flinkFs =
 	org.apache.flink.core.fs.FileSystem.get(path.toUri());
final FileSystem hadoopFs = (flinkFs instanceof HadoopFileSystem)
	? ((HadoopFileSystem) flinkFs).getHadoopFileSystem()
	: null;

FileSystem#get() wraps the created FileSystem in a SafetyNetWrapperFileSystem, resulting in the instanceof check to categorically fail.

We may want to replace the get() call with getUnguardedFileSystem(). We should also look for other occurrences of similar instanceof checks.

According to a thread on the mailing list this causes the BucketingSink to be unusable. http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/BucketingSink-broken-in-flink-1-4-0-td17710.html