[FLINK-2555] Properly pass security credentials in the Hadoop Input/Output format wrappers

[rmetzger]

This is needed because the Hadoop IF/OF's are using Hadoop's FileSystem stack, which is using the security credentials passed in the JobConf / Job class in the getSplits() method.

Note that access to secured Hadoop 1.x using Hadoop IF/OF's is not possible with this change. This limitation is due to missing methods in the old APIs.

I've also updated the version of the "de.javakaffee.kryo-serializers" from 0.27 to 0.36 because a user on the ML recently needed a specific Kryo serializer which was not available in the old dependency.

For the Java and Scala API, I renamed the first argument's name: readHadoopFile(org.apache.hadoop.mapreduce.lib.input.FileInputFormat<K,V> mapreduceInputFormat, Class<K> key, Class<V> value, String inputPath, Job job)

This makes it easier in IDE completions to distinguish between the mapreduce and the mapred variant. (before the argument was always called mapredInputFormat now, we have the mapreduceInputFormat variant where applicable)

[StephanEwen]

Looks good. The HadoopFormatBase and similar classes could use a line or two more in comments, but otherwise, this seems well.

Any way to test this? There does not seem to be any test for the format wrappers, yet...

[rmetzger]

I actually think that there is no need for the HadoopInputFormatBases to exist.
There are two implementations and two bases for mapred and mapreduce, but they have nothing in common.

There are some tests for the non secure case in org.apache.flink.test.hadoop.

[rmetzger]

There might actually be a way of testing against a secured cluster: https://issues.apache.org/jira/browse/HADOOP-9848 / https://github.com/apache/hadoop/blob/master/hadoop-common-project/hadoop-minikdc/src/main/java/org/apache/hadoop/minikdc/MiniKdc.java
This seems to be available since Hadoop 2.3.0

[aljoscha]

The Bases exist because there is a java-specific and a scals-specific version of each HadoopInputFormat.

[rmetzger]

Okay, that makes sense.
I'll add some comments to the classes.

[rmetzger]

@mxm: I removed the comment.

[mxm]

It would be great if we implemented a test case against the MiniKDC server.

[rmetzger]

I agree. Lets file a JIRA and do it separately, as this is probably a bigger task.

[mxm]

I've opened another issue for that: https://issues.apache.org/jira/browse/FLINK-2573

[uce]

whitespace

[uce]

I'll address my trivial comment and merge this. Thanks!

[rmetzger]

Thanks alot!

[rmetzger]

I'm manually closing this pull request. It has been merged by @uce.