flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From aljos...@apache.org
Subject [07/20] flink git commit: [FLINK-4048] Remove Hadoop from DataSet API
Date Wed, 27 Sep 2017 11:09:12 GMT
http://git-wip-us.apache.org/repos/asf/flink/blob/4beff13e/flink-connectors/flink-hadoop-compatibility/src/main/scala/org/apache/flink/api/scala/hadoop/mapred/HadoopInputFormat.scala
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-hadoop-compatibility/src/main/scala/org/apache/flink/api/scala/hadoop/mapred/HadoopInputFormat.scala b/flink-connectors/flink-hadoop-compatibility/src/main/scala/org/apache/flink/api/scala/hadoop/mapred/HadoopInputFormat.scala
new file mode 100644
index 0000000..52a6bed
--- /dev/null
+++ b/flink-connectors/flink-hadoop-compatibility/src/main/scala/org/apache/flink/api/scala/hadoop/mapred/HadoopInputFormat.scala
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.api.scala.hadoop.mapred
+
+import org.apache.flink.annotation.Public
+import org.apache.flink.api.java.hadoop.mapred.HadoopInputFormatBase
+import org.apache.hadoop.mapred.{InputFormat, JobConf}
+
+@Public
+class HadoopInputFormat[K, V](
+    mapredInputFormat: InputFormat[K, V],
+    keyClass: Class[K],
+    valueClass: Class[V],
+    job: JobConf)
+  extends HadoopInputFormatBase[K, V, (K, V)](mapredInputFormat, keyClass, valueClass, job) {
+
+  def this(mapredInputFormat: InputFormat[K, V], keyClass: Class[K], valueClass: Class[V]) = {
+    this(mapredInputFormat, keyClass, valueClass, new JobConf)
+  }
+
+  def nextRecord(reuse: (K, V)): (K, V) = {
+    if (!fetched) {
+      fetchNext()
+    }
+    if (!hasNext) {
+      null
+    } else {
+      fetched = false
+      (key, value)
+    }
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/4beff13e/flink-connectors/flink-hadoop-compatibility/src/main/scala/org/apache/flink/api/scala/hadoop/mapred/HadoopOutputFormat.scala
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-hadoop-compatibility/src/main/scala/org/apache/flink/api/scala/hadoop/mapred/HadoopOutputFormat.scala b/flink-connectors/flink-hadoop-compatibility/src/main/scala/org/apache/flink/api/scala/hadoop/mapred/HadoopOutputFormat.scala
new file mode 100644
index 0000000..b786651
--- /dev/null
+++ b/flink-connectors/flink-hadoop-compatibility/src/main/scala/org/apache/flink/api/scala/hadoop/mapred/HadoopOutputFormat.scala
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.api.scala.hadoop.mapred
+
+import org.apache.flink.annotation.Public
+import org.apache.flink.api.java.hadoop.mapred.HadoopOutputFormatBase
+import org.apache.hadoop.mapred.{JobConf, OutputCommitter, OutputFormat}
+
+@Public
+class HadoopOutputFormat[K, V](mapredOutputFormat: OutputFormat[K, V], job: JobConf)
+  extends HadoopOutputFormatBase[K, V, (K, V)](mapredOutputFormat, job) {
+
+  def this(
+      mapredOutputFormat: OutputFormat[K, V],
+      outputCommitterClass: Class[OutputCommitter],
+      job: JobConf) {
+
+    this(mapredOutputFormat, job)
+    this.getJobConf.setOutputCommitter(outputCommitterClass)
+  }
+
+  def writeRecord(record: (K, V)) {
+    this.recordWriter.write(record._1, record._2)
+  }
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/4beff13e/flink-connectors/flink-hadoop-compatibility/src/main/scala/org/apache/flink/api/scala/hadoop/mapreduce/HadoopInputFormat.scala
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-hadoop-compatibility/src/main/scala/org/apache/flink/api/scala/hadoop/mapreduce/HadoopInputFormat.scala b/flink-connectors/flink-hadoop-compatibility/src/main/scala/org/apache/flink/api/scala/hadoop/mapreduce/HadoopInputFormat.scala
new file mode 100644
index 0000000..8efb53d
--- /dev/null
+++ b/flink-connectors/flink-hadoop-compatibility/src/main/scala/org/apache/flink/api/scala/hadoop/mapreduce/HadoopInputFormat.scala
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.scala.hadoop.mapreduce
+
+import org.apache.flink.annotation.Public
+import org.apache.flink.api.java.hadoop.mapreduce.HadoopInputFormatBase
+import org.apache.hadoop.mapreduce.{InputFormat, Job}
+
+@Public
+class HadoopInputFormat[K, V](
+    mapredInputFormat: InputFormat[K, V],
+    keyClass: Class[K],
+    valueClass: Class[V],
+    job: Job)
+  extends HadoopInputFormatBase[K, V, (K, V)](mapredInputFormat, keyClass, valueClass, job) {
+
+  def this(mapredInputFormat: InputFormat[K, V], keyClass: Class[K], valueClass: Class[V]) = {
+    this(mapredInputFormat, keyClass, valueClass, Job.getInstance())
+  }
+
+  def nextRecord(reuse: (K, V)): (K, V) = {
+    if (!fetched) {
+      fetchNext()
+    }
+    if (!hasNext) {
+      null
+    } else {
+      fetched = false
+      (recordReader.getCurrentKey, recordReader.getCurrentValue)
+    }
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/4beff13e/flink-connectors/flink-hadoop-compatibility/src/main/scala/org/apache/flink/api/scala/hadoop/mapreduce/HadoopOutputFormat.scala
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-hadoop-compatibility/src/main/scala/org/apache/flink/api/scala/hadoop/mapreduce/HadoopOutputFormat.scala b/flink-connectors/flink-hadoop-compatibility/src/main/scala/org/apache/flink/api/scala/hadoop/mapreduce/HadoopOutputFormat.scala
new file mode 100644
index 0000000..8095304
--- /dev/null
+++ b/flink-connectors/flink-hadoop-compatibility/src/main/scala/org/apache/flink/api/scala/hadoop/mapreduce/HadoopOutputFormat.scala
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.scala.hadoop.mapreduce
+
+import org.apache.flink.annotation.Public
+import org.apache.flink.api.java.hadoop.mapreduce.HadoopOutputFormatBase
+import org.apache.hadoop.mapreduce.{Job, OutputFormat}
+
+@Public
+class HadoopOutputFormat[K, V](mapredOutputFormat: OutputFormat[K, V], job: Job)
+  extends HadoopOutputFormatBase[K, V, (K, V)](mapredOutputFormat, job) {
+
+  def writeRecord(record: (K, V)) {
+    this.recordWriter.write(record._1, record._2)
+  }
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/4beff13e/flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormatTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormatTest.java b/flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormatTest.java
new file mode 100644
index 0000000..bfee273
--- /dev/null
+++ b/flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormatTest.java
@@ -0,0 +1,307 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.java.hadoop.mapred;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.hadoop.mapred.wrapper.HadoopInputSplit;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+
+import org.apache.hadoop.conf.Configurable;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.mapred.FileInputFormat;
+import org.apache.hadoop.mapred.FileSplit;
+import org.apache.hadoop.mapred.InputFormat;
+import org.apache.hadoop.mapred.InputSplit;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.JobConfigurable;
+import org.apache.hadoop.mapred.RecordReader;
+import org.apache.hadoop.mapred.Reporter;
+import org.junit.Test;
+
+import java.io.IOException;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.core.Is.is;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.anyInt;
+import static org.mockito.Mockito.anyLong;
+import static org.mockito.Mockito.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for {@link HadoopInputFormat}.
+ */
+public class HadoopInputFormatTest {
+
+	@Test
+	public void testConfigureWithConfigurableInstance() {
+		ConfigurableDummyInputFormat inputFormat = mock(ConfigurableDummyInputFormat.class);
+
+		HadoopInputFormat<String, Long> hadoopInputFormat = new HadoopInputFormat<>(inputFormat, String.class, Long.class, new JobConf());
+		verify(inputFormat, times(1)).setConf(any(JobConf.class));
+
+		hadoopInputFormat.configure(new org.apache.flink.configuration.Configuration());
+		verify(inputFormat, times(2)).setConf(any(JobConf.class));
+	}
+
+	@Test
+	public void testConfigureWithJobConfigurableInstance() {
+		JobConfigurableDummyInputFormat inputFormat = mock(JobConfigurableDummyInputFormat.class);
+
+		HadoopInputFormat<String, Long> hadoopInputFormat = new HadoopInputFormat<>(inputFormat, String.class, Long.class, new JobConf());
+		verify(inputFormat, times(1)).configure(any(JobConf.class));
+
+		hadoopInputFormat.configure(new org.apache.flink.configuration.Configuration());
+		verify(inputFormat, times(2)).configure(any(JobConf.class));
+	}
+
+	@Test
+	public void testOpenClose() throws Exception {
+		DummyRecordReader recordReader = mock(DummyRecordReader.class);
+		DummyInputFormat inputFormat = mock(DummyInputFormat.class);
+		when(inputFormat.getRecordReader(any(InputSplit.class), any(JobConf.class), any(Reporter.class))).thenReturn(recordReader);
+
+		HadoopInputFormat<String, Long> hadoopInputFormat = new HadoopInputFormat<>(inputFormat, String.class, Long.class, new JobConf());
+		hadoopInputFormat.open(getHadoopInputSplit());
+
+		verify(inputFormat, times(1)).getRecordReader(any(InputSplit.class), any(JobConf.class), any(Reporter.class));
+		verify(recordReader, times(1)).createKey();
+		verify(recordReader, times(1)).createValue();
+
+		assertThat(hadoopInputFormat.fetched, is(false));
+
+		hadoopInputFormat.close();
+		verify(recordReader, times(1)).close();
+	}
+
+	@Test
+	public void testOpenWithConfigurableReader() throws Exception {
+		ConfigurableDummyRecordReader recordReader = mock(ConfigurableDummyRecordReader.class);
+		DummyInputFormat inputFormat = mock(DummyInputFormat.class);
+		when(inputFormat.getRecordReader(any(InputSplit.class), any(JobConf.class), any(Reporter.class))).thenReturn(recordReader);
+
+		HadoopInputFormat<String, Long> hadoopInputFormat = new HadoopInputFormat<>(inputFormat, String.class, Long.class, new JobConf());
+		hadoopInputFormat.open(getHadoopInputSplit());
+
+		verify(inputFormat, times(1)).getRecordReader(any(InputSplit.class), any(JobConf.class), any(Reporter.class));
+		verify(recordReader, times(1)).setConf(any(JobConf.class));
+		verify(recordReader, times(1)).createKey();
+		verify(recordReader, times(1)).createValue();
+
+		assertThat(hadoopInputFormat.fetched, is(false));
+
+	}
+
+	@Test
+	public void testCreateInputSplits() throws Exception {
+
+		FileSplit[] result = new FileSplit[1];
+		result[0] = getFileSplit();
+		DummyInputFormat inputFormat = mock(DummyInputFormat.class);
+		when(inputFormat.getSplits(any(JobConf.class), anyInt())).thenReturn(result);
+
+		HadoopInputFormat<String, Long> hadoopInputFormat = new HadoopInputFormat<>(inputFormat, String.class, Long.class, new JobConf());
+		hadoopInputFormat.createInputSplits(2);
+
+		verify(inputFormat, times(1)).getSplits(any(JobConf.class), anyInt());
+	}
+
+	@Test
+	public void testReachedEndWithElementsRemaining() throws IOException {
+
+		HadoopInputFormat<String, Long> hadoopInputFormat = new HadoopInputFormat<>(new DummyInputFormat(), String.class, Long.class, new JobConf());
+		hadoopInputFormat.fetched = true;
+		hadoopInputFormat.hasNext = true;
+
+		assertThat(hadoopInputFormat.reachedEnd(), is(false));
+	}
+
+	@Test
+	public void testReachedEndWithNoElementsRemaining() throws IOException {
+		HadoopInputFormat<String, Long> hadoopInputFormat = new HadoopInputFormat<>(new DummyInputFormat(), String.class, Long.class, new JobConf());
+		hadoopInputFormat.fetched = true;
+		hadoopInputFormat.hasNext = false;
+
+		assertThat(hadoopInputFormat.reachedEnd(), is(true));
+	}
+
+	@Test
+	public void testFetchNext() throws IOException {
+		DummyRecordReader recordReader = mock(DummyRecordReader.class);
+		when(recordReader.next(anyString(), anyLong())).thenReturn(true);
+
+		DummyInputFormat inputFormat = mock(DummyInputFormat.class);
+		when(inputFormat.getRecordReader(any(InputSplit.class), any(JobConf.class), any(Reporter.class))).thenReturn(recordReader);
+
+		HadoopInputFormat<String, Long> hadoopInputFormat = new HadoopInputFormat<>(inputFormat, String.class, Long.class, new JobConf());
+		hadoopInputFormat.open(getHadoopInputSplit());
+		hadoopInputFormat.fetchNext();
+
+		verify(recordReader, times(1)).next(anyString(), anyLong());
+		assertThat(hadoopInputFormat.hasNext, is(true));
+		assertThat(hadoopInputFormat.fetched, is(true));
+	}
+
+	@Test
+	public void checkTypeInformation() throws Exception {
+		HadoopInputFormat<Void, Long> hadoopInputFormat = new HadoopInputFormat<>(
+				new DummyVoidKeyInputFormat<Long>(), Void.class, Long.class, new JobConf());
+
+		TypeInformation<Tuple2<Void, Long>> tupleType = hadoopInputFormat.getProducedType();
+		TypeInformation<Tuple2<Void, Long>> expectedType = new TupleTypeInfo<>(BasicTypeInfo.VOID_TYPE_INFO, BasicTypeInfo.LONG_TYPE_INFO);
+
+		assertThat(tupleType.isTupleType(), is(true));
+		assertThat(tupleType, is(equalTo(expectedType)));
+	}
+
+	@Test
+	public void testCloseWithoutOpen() throws Exception {
+		HadoopInputFormat<Void, Long> hadoopInputFormat = new HadoopInputFormat<>(
+			new DummyVoidKeyInputFormat<Long>(), Void.class, Long.class, new JobConf());
+		hadoopInputFormat.close();
+	}
+
+	private HadoopInputSplit getHadoopInputSplit() {
+		return new HadoopInputSplit(1, getFileSplit(), new JobConf());
+	}
+
+	private FileSplit getFileSplit() {
+		return new FileSplit(new Path("path"), 1, 2, new String[]{});
+	}
+
+	private class DummyVoidKeyInputFormat<T> extends FileInputFormat<Void, T> {
+
+		public DummyVoidKeyInputFormat() {}
+
+		@Override
+		public org.apache.hadoop.mapred.RecordReader<Void, T> getRecordReader(org.apache.hadoop.mapred.InputSplit inputSplit, JobConf jobConf, Reporter reporter) throws IOException {
+			return null;
+		}
+	}
+
+	private class DummyRecordReader implements RecordReader<String, Long> {
+
+		@Override
+		public float getProgress() throws IOException {
+			return 0;
+		}
+
+		@Override
+		public boolean next(String s, Long aLong) throws IOException {
+			return false;
+		}
+
+		@Override
+		public String createKey() {
+			return null;
+		}
+
+		@Override
+		public Long createValue() {
+			return null;
+		}
+
+		@Override
+		public long getPos() throws IOException {
+			return 0;
+		}
+
+		@Override
+		public void close() throws IOException {
+
+		}
+	}
+
+	private class ConfigurableDummyRecordReader implements RecordReader<String, Long>, Configurable {
+
+		@Override
+		public void setConf(Configuration configuration) {}
+
+		@Override
+		public Configuration getConf() {
+			return null;
+		}
+
+		@Override
+		public boolean next(String s, Long aLong) throws IOException {
+			return false;
+		}
+
+		@Override
+		public String createKey() {
+			return null;
+		}
+
+		@Override
+		public Long createValue() {
+			return null;
+		}
+
+		@Override
+		public long getPos() throws IOException {
+			return 0;
+		}
+
+		@Override
+		public void close() throws IOException {
+
+		}
+
+		@Override
+		public float getProgress() throws IOException {
+			return 0;
+		}
+	}
+
+	private class DummyInputFormat implements InputFormat<String, Long> {
+
+		@Override
+		public InputSplit[] getSplits(JobConf jobConf, int i) throws IOException {
+			return new InputSplit[0];
+		}
+
+		@Override
+		public RecordReader<String, Long> getRecordReader(InputSplit inputSplit, JobConf jobConf, Reporter reporter) throws IOException {
+			return null;
+		}
+	}
+
+	private class ConfigurableDummyInputFormat extends DummyInputFormat implements Configurable {
+		@Override
+		public void setConf(Configuration configuration) {}
+
+		@Override
+		public Configuration getConf() {
+			return null;
+		}
+	}
+
+	private class JobConfigurableDummyInputFormat extends DummyInputFormat implements JobConfigurable {
+
+		@Override
+		public void configure(JobConf jobConf) {}
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/4beff13e/flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/api/java/hadoop/mapred/HadoopOutputFormatTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/api/java/hadoop/mapred/HadoopOutputFormatTest.java b/flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/api/java/hadoop/mapred/HadoopOutputFormatTest.java
new file mode 100644
index 0000000..362a37f
--- /dev/null
+++ b/flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/api/java/hadoop/mapred/HadoopOutputFormatTest.java
@@ -0,0 +1,233 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.java.hadoop.mapred;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+
+import org.apache.hadoop.conf.Configurable;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.JobConfigurable;
+import org.apache.hadoop.mapred.JobContext;
+import org.apache.hadoop.mapred.OutputCommitter;
+import org.apache.hadoop.mapred.OutputFormat;
+import org.apache.hadoop.mapred.RecordWriter;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.mapred.TaskAttemptContext;
+import org.apache.hadoop.util.Progressable;
+import org.junit.Test;
+import org.mockito.Matchers;
+import org.mockito.Mockito;
+
+import java.io.IOException;
+
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.anyLong;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for {@link HadoopOutputFormat}.
+ */
+public class HadoopOutputFormatTest {
+
+	@Test
+	public void testOpen() throws Exception {
+
+		OutputFormat<String, Long> dummyOutputFormat = mock(DummyOutputFormat.class);
+		DummyOutputCommitter outputCommitter = mock(DummyOutputCommitter.class);
+		JobConf jobConf = Mockito.spy(new JobConf());
+		when(jobConf.getOutputCommitter()).thenReturn(outputCommitter);
+
+		HadoopOutputFormat<String, Long> outputFormat = new HadoopOutputFormat<>(dummyOutputFormat, jobConf);
+
+		outputFormat.open(1, 1);
+
+		verify(jobConf, times(2)).getOutputCommitter();
+		verify(outputCommitter, times(1)).setupJob(any(JobContext.class));
+		verify(dummyOutputFormat, times(1)).getRecordWriter(any(FileSystem.class), any(JobConf.class), anyString(), any(Progressable.class));
+	}
+
+	@Test
+	public void testConfigureWithConfigurable() {
+		ConfigurableDummyOutputFormat dummyOutputFormat = mock(ConfigurableDummyOutputFormat.class);
+		JobConf jobConf = mock(JobConf.class);
+
+		HadoopOutputFormat<String, Long> outputFormat = new HadoopOutputFormat<>(dummyOutputFormat, jobConf);
+
+		outputFormat.configure(Matchers.<org.apache.flink.configuration.Configuration>any());
+
+		verify(dummyOutputFormat, times(1)).setConf(any(Configuration.class));
+	}
+
+	@Test
+	public void testConfigureWithJobConfigurable() {
+		JobConfigurableDummyOutputFormat dummyOutputFormat = mock(JobConfigurableDummyOutputFormat.class);
+		JobConf jobConf = mock(JobConf.class);
+
+		HadoopOutputFormat<String, Long> outputFormat = new HadoopOutputFormat<>(dummyOutputFormat, jobConf);
+
+		outputFormat.configure(Matchers.<org.apache.flink.configuration.Configuration>any());
+
+		verify(dummyOutputFormat, times(1)).configure(any(JobConf.class));
+	}
+
+	@Test
+	public void testCloseWithTaskCommit() throws Exception {
+		OutputFormat<String, Long> dummyOutputFormat = mock(DummyOutputFormat.class);
+		DummyOutputCommitter outputCommitter = mock(DummyOutputCommitter.class);
+		when(outputCommitter.needsTaskCommit(any(TaskAttemptContext.class))).thenReturn(true);
+		DummyRecordWriter recordWriter = mock(DummyRecordWriter.class);
+		JobConf jobConf = mock(JobConf.class);
+
+		HadoopOutputFormat<String, Long> outputFormat = new HadoopOutputFormat<>(dummyOutputFormat, jobConf);
+		outputFormat.recordWriter = recordWriter;
+		outputFormat.outputCommitter = outputCommitter;
+
+		outputFormat.close();
+
+		verify(recordWriter, times(1)).close(any(Reporter.class));
+		verify(outputCommitter, times(1)).commitTask(any(TaskAttemptContext.class));
+	}
+
+	@Test
+	public void testCloseWithoutTaskCommit() throws Exception {
+		OutputFormat<String, Long> dummyOutputFormat = mock(DummyOutputFormat.class);
+		DummyOutputCommitter outputCommitter = mock(DummyOutputCommitter.class);
+		when(outputCommitter.needsTaskCommit(any(TaskAttemptContext.class))).thenReturn(false);
+		DummyRecordWriter recordWriter = mock(DummyRecordWriter.class);
+		JobConf jobConf = mock(JobConf.class);
+
+		HadoopOutputFormat<String, Long> outputFormat = new HadoopOutputFormat<>(dummyOutputFormat, jobConf);
+		outputFormat.recordWriter = recordWriter;
+		outputFormat.outputCommitter = outputCommitter;
+
+		outputFormat.close();
+
+		verify(recordWriter, times(1)).close(any(Reporter.class));
+		verify(outputCommitter, times(0)).commitTask(any(TaskAttemptContext.class));
+	}
+
+	@Test
+	public void testWriteRecord() throws Exception {
+		OutputFormat<String, Long> dummyOutputFormat = mock(DummyOutputFormat.class);
+		DummyRecordWriter recordWriter = mock(DummyRecordWriter.class);
+		JobConf jobConf = mock(JobConf.class);
+
+		HadoopOutputFormat<String, Long> outputFormat = new HadoopOutputFormat<>(dummyOutputFormat, jobConf);
+		outputFormat.recordWriter = recordWriter;
+
+		outputFormat.writeRecord(new Tuple2<>("key", 1L));
+
+		verify(recordWriter, times(1)).write(anyString(), anyLong());
+	}
+
+	@Test
+	public void testFinalizeGlobal() throws Exception {
+		OutputFormat<String, Long> dummyOutputFormat = mock(DummyOutputFormat.class);
+		DummyOutputCommitter outputCommitter = mock(DummyOutputCommitter.class);
+		JobConf jobConf = Mockito.spy(new JobConf());
+		when(jobConf.getOutputCommitter()).thenReturn(outputCommitter);
+
+		HadoopOutputFormat<String, Long> outputFormat = new HadoopOutputFormat<>(dummyOutputFormat, jobConf);
+
+		outputFormat.finalizeGlobal(1);
+
+		verify(outputCommitter, times(1)).commitJob(any(JobContext.class));
+	}
+
+	private class DummyOutputFormat implements OutputFormat<String, Long> {
+
+		@Override
+		public RecordWriter<String, Long> getRecordWriter(FileSystem fileSystem, JobConf jobConf, String s, Progressable progressable) throws IOException {
+			return null;
+		}
+
+		@Override
+		public void checkOutputSpecs(FileSystem fileSystem, JobConf jobConf) throws IOException {
+
+		}
+	}
+
+	private class ConfigurableDummyOutputFormat extends DummyOutputFormat implements Configurable {
+
+		@Override
+		public void setConf(Configuration configuration) {
+
+		}
+
+		@Override
+		public Configuration getConf() {
+			return null;
+		}
+	}
+
+	private class JobConfigurableDummyOutputFormat extends DummyOutputFormat implements JobConfigurable {
+
+		@Override
+		public void configure(JobConf jobConf) {
+
+		}
+	}
+
+	private class DummyOutputCommitter extends OutputCommitter {
+
+		@Override
+		public void setupJob(JobContext jobContext) throws IOException {
+
+		}
+
+		@Override
+		public void setupTask(TaskAttemptContext taskAttemptContext) throws IOException {
+
+		}
+
+		@Override
+		public boolean needsTaskCommit(TaskAttemptContext taskAttemptContext) throws IOException {
+			return false;
+		}
+
+		@Override
+		public void commitTask(TaskAttemptContext taskAttemptContext) throws IOException {
+
+		}
+
+		@Override
+		public void abortTask(TaskAttemptContext taskAttemptContext) throws IOException {
+
+		}
+	}
+
+	private class DummyRecordWriter implements RecordWriter<String, Long> {
+
+		@Override
+		public void write(String s, Long aLong) throws IOException {
+
+		}
+
+		@Override
+		public void close(Reporter reporter) throws IOException {
+
+		}
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/4beff13e/flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormatTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormatTest.java b/flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormatTest.java
new file mode 100644
index 0000000..941eb1f
--- /dev/null
+++ b/flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormatTest.java
@@ -0,0 +1,238 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.java.hadoop.mapreduce;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.hadoop.mapreduce.wrapper.HadoopInputSplit;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+
+import org.apache.hadoop.conf.Configurable;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapreduce.InputFormat;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.JobContext;
+import org.apache.hadoop.mapreduce.RecordReader;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+import java.io.IOException;
+import java.util.List;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.core.Is.is;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for {@link HadoopInputFormat}.
+ */
+public class HadoopInputFormatTest {
+
+	@Rule
+	public final ExpectedException exception = ExpectedException.none();
+
+	@Test
+	public void testConfigure() throws Exception {
+
+		ConfigurableDummyInputFormat inputFormat = mock(ConfigurableDummyInputFormat.class);
+
+		HadoopInputFormat<String, Long> hadoopInputFormat = setupHadoopInputFormat(inputFormat, Job.getInstance(), null);
+		hadoopInputFormat.configure(new org.apache.flink.configuration.Configuration());
+
+		verify(inputFormat, times(1)).setConf(any(Configuration.class));
+	}
+
+	@Test
+	public void testCreateInputSplits() throws Exception {
+		DummyInputFormat inputFormat = mock(DummyInputFormat.class);
+
+		HadoopInputFormat<String, Long> hadoopInputFormat = setupHadoopInputFormat(inputFormat, Job.getInstance(), null);
+		hadoopInputFormat.createInputSplits(2);
+
+		verify(inputFormat, times(1)).getSplits(any(JobContext.class));
+	}
+
+	@Test
+	public void testOpen() throws Exception {
+		DummyInputFormat inputFormat = mock(DummyInputFormat.class);
+		when(inputFormat.createRecordReader(any(InputSplit.class), any(TaskAttemptContext.class))).thenReturn(new DummyRecordReader());
+		HadoopInputSplit inputSplit = mock(HadoopInputSplit.class);
+
+		HadoopInputFormat<String, Long> hadoopInputFormat = setupHadoopInputFormat(inputFormat, Job.getInstance(), null);
+		hadoopInputFormat.open(inputSplit);
+
+		verify(inputFormat, times(1)).createRecordReader(any(InputSplit.class), any(TaskAttemptContext.class));
+		assertThat(hadoopInputFormat.fetched, is(false));
+	}
+
+	@Test
+	public void testClose() throws Exception {
+
+		DummyRecordReader recordReader = mock(DummyRecordReader.class);
+
+		HadoopInputFormat<String, Long> hadoopInputFormat = setupHadoopInputFormat(new DummyInputFormat(), Job.getInstance(), recordReader);
+		hadoopInputFormat.close();
+
+		verify(recordReader, times(1)).close();
+	}
+
+	@Test
+	public void testCloseWithoutOpen() throws Exception {
+		HadoopInputFormat<String, Long> hadoopInputFormat = new HadoopInputFormat<>(new DummyInputFormat(), String.class, Long.class, Job.getInstance());
+		hadoopInputFormat.close();
+	}
+
+	@Test
+	public void testFetchNextInitialState() throws Exception {
+		DummyRecordReader recordReader = new DummyRecordReader();
+
+		HadoopInputFormat<String, Long> hadoopInputFormat = setupHadoopInputFormat(new DummyInputFormat(), Job.getInstance(), recordReader);
+		hadoopInputFormat.fetchNext();
+
+		assertThat(hadoopInputFormat.fetched, is(true));
+		assertThat(hadoopInputFormat.hasNext, is(false));
+	}
+
+	@Test
+	public void testFetchNextRecordReaderHasNewValue() throws Exception {
+
+		DummyRecordReader recordReader = mock(DummyRecordReader.class);
+		when(recordReader.nextKeyValue()).thenReturn(true);
+
+		HadoopInputFormat<String, Long> hadoopInputFormat = setupHadoopInputFormat(new DummyInputFormat(), Job.getInstance(), recordReader);
+		hadoopInputFormat.fetchNext();
+
+		assertThat(hadoopInputFormat.fetched, is(true));
+		assertThat(hadoopInputFormat.hasNext, is(true));
+	}
+
+	@Test
+	public void testFetchNextRecordReaderThrowsException() throws Exception {
+
+		DummyRecordReader recordReader = mock(DummyRecordReader.class);
+		when(recordReader.nextKeyValue()).thenThrow(new InterruptedException());
+
+		HadoopInputFormat<String, Long> hadoopInputFormat = setupHadoopInputFormat(new DummyInputFormat(), Job.getInstance(), recordReader);
+
+		exception.expect(IOException.class);
+		hadoopInputFormat.fetchNext();
+
+		assertThat(hadoopInputFormat.hasNext, is(true));
+	}
+
+	@Test
+	public void checkTypeInformation() throws Exception {
+
+		HadoopInputFormat<Void, Long> hadoopInputFormat = new HadoopInputFormat<>(
+				new DummyVoidKeyInputFormat<Long>(), Void.class, Long.class, Job.getInstance());
+
+		TypeInformation<Tuple2<Void, Long>> tupleType = hadoopInputFormat.getProducedType();
+		TypeInformation<Tuple2<Void, Long>> expectedType = new TupleTypeInfo<>(BasicTypeInfo.VOID_TYPE_INFO, BasicTypeInfo.LONG_TYPE_INFO);
+
+		assertThat(tupleType.isTupleType(), is(true));
+		assertThat(tupleType, is(equalTo(expectedType)));
+	}
+
+	private HadoopInputFormat<String, Long> setupHadoopInputFormat(InputFormat<String, Long> inputFormat, Job job,
+																	RecordReader<String, Long> recordReader) {
+
+		HadoopInputFormat<String, Long> hadoopInputFormat = new HadoopInputFormat<>(inputFormat,
+				String.class, Long.class, job);
+		hadoopInputFormat.recordReader = recordReader;
+
+		return hadoopInputFormat;
+	}
+
+	private class DummyVoidKeyInputFormat<T> extends FileInputFormat<Void, T> {
+
+		public DummyVoidKeyInputFormat() {}
+
+		@Override
+		public RecordReader<Void, T> createRecordReader(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
+			return null;
+		}
+	}
+
+	private class DummyRecordReader extends RecordReader<String, Long> {
+
+		@Override
+		public void initialize(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
+
+		}
+
+		@Override
+		public boolean nextKeyValue() throws IOException, InterruptedException {
+			return false;
+		}
+
+		@Override
+		public String getCurrentKey() throws IOException, InterruptedException {
+			return null;
+		}
+
+		@Override
+		public Long getCurrentValue() throws IOException, InterruptedException {
+			return null;
+		}
+
+		@Override
+		public float getProgress() throws IOException, InterruptedException {
+			return 0;
+		}
+
+		@Override
+		public void close() throws IOException {
+
+		}
+	}
+
+	private class DummyInputFormat extends InputFormat<String, Long> {
+
+		@Override
+		public List<InputSplit> getSplits(JobContext jobContext) throws IOException, InterruptedException {
+			return null;
+		}
+
+		@Override
+		public RecordReader<String, Long> createRecordReader(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
+			return new DummyRecordReader();
+		}
+	}
+
+	private class ConfigurableDummyInputFormat extends DummyInputFormat implements Configurable {
+
+		@Override
+		public void setConf(Configuration configuration) {}
+
+		@Override
+		public Configuration getConf() {
+			return null;
+		}
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/4beff13e/flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopOutputFormatTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopOutputFormatTest.java b/flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopOutputFormatTest.java
new file mode 100644
index 0000000..8e2329a
--- /dev/null
+++ b/flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopOutputFormatTest.java
@@ -0,0 +1,200 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.java.hadoop.mapreduce;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+
+import org.apache.hadoop.conf.Configurable;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.JobContext;
+import org.apache.hadoop.mapreduce.OutputCommitter;
+import org.apache.hadoop.mapreduce.OutputFormat;
+import org.apache.hadoop.mapreduce.RecordWriter;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.io.IOException;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.anyLong;
+import static org.mockito.Mockito.anyString;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for {@link HadoopOutputFormat}.
+ */
+public class HadoopOutputFormatTest {
+
+	private static final String MAPRED_OUTPUT_PATH = "an/ignored/file/";
+	private static final String MAPRED_OUTPUT_DIR_KEY = "mapred.output.dir";
+
+	@Test
+	public void testWriteRecord() throws Exception {
+
+		RecordWriter<String, Long> recordWriter = mock(DummyRecordWriter.class);
+		HadoopOutputFormat<String, Long> hadoopOutputFormat = setupHadoopOutputFormat(new DummyOutputFormat(),
+			Job.getInstance(), recordWriter, null, new Configuration());
+
+		hadoopOutputFormat.writeRecord(new Tuple2<String, Long>());
+
+		verify(recordWriter, times(1)).write(anyString(), anyLong());
+	}
+
+	@Test
+	public void testOpen() throws Exception {
+
+		OutputFormat<String, Long> dummyOutputFormat = mock(DummyOutputFormat.class);
+		OutputCommitter outputCommitter = setupOutputCommitter(true);
+		when(dummyOutputFormat.getOutputCommitter(any(TaskAttemptContext.class))).thenReturn(outputCommitter);
+
+		HadoopOutputFormat<String, Long> hadoopOutputFormat = setupHadoopOutputFormat(dummyOutputFormat,
+			Job.getInstance(), new DummyRecordWriter(), setupOutputCommitter(true), new Configuration());
+
+		hadoopOutputFormat.open(1, 4);
+
+		verify(hadoopOutputFormat.outputCommitter, times(1)).setupJob(any(JobContext.class));
+		verify(hadoopOutputFormat.mapreduceOutputFormat, times(1)).getRecordWriter(any(TaskAttemptContext.class));
+	}
+
+	@Test
+	public void testCloseWithNeedsTaskCommitTrue() throws Exception {
+
+		RecordWriter<String, Long> recordWriter = Mockito.mock(DummyRecordWriter.class);
+		OutputCommitter outputCommitter = setupOutputCommitter(true);
+
+		HadoopOutputFormat<String, Long> hadoopOutputFormat = setupHadoopOutputFormat(new DummyOutputFormat(),
+			Job.getInstance(), recordWriter, outputCommitter, new Configuration());
+
+		hadoopOutputFormat.close();
+
+		verify(outputCommitter, times(1)).commitTask(any(TaskAttemptContext.class));
+		verify(recordWriter, times(1)).close(any(TaskAttemptContext.class));
+	}
+
+	@Test
+	public void testCloseWithNeedsTaskCommitFalse() throws Exception {
+
+		RecordWriter<String, Long> recordWriter = Mockito.mock(DummyRecordWriter.class);
+		OutputCommitter outputCommitter = setupOutputCommitter(false);
+
+		HadoopOutputFormat<String, Long> hadoopOutputFormat = setupHadoopOutputFormat(new DummyOutputFormat(),
+			Job.getInstance(), recordWriter, outputCommitter, new Configuration());
+
+		hadoopOutputFormat.close();
+
+		verify(outputCommitter, times(0)).commitTask(any(TaskAttemptContext.class));
+		verify(recordWriter, times(1)).close(any(TaskAttemptContext.class));
+	}
+
+	@Test
+	public void testConfigure() throws Exception {
+
+		ConfigurableDummyOutputFormat outputFormat = mock(ConfigurableDummyOutputFormat.class);
+
+		HadoopOutputFormat<String, Long> hadoopOutputFormat = setupHadoopOutputFormat(outputFormat, Job.getInstance(),
+			null, null, new Configuration());
+
+		hadoopOutputFormat.configure(new org.apache.flink.configuration.Configuration());
+
+		verify(outputFormat, times(1)).setConf(any(Configuration.class));
+	}
+
+	@Test
+	public void testFinalizedGlobal() throws Exception {
+
+		HadoopOutputFormat<String, Long> hadoopOutputFormat = setupHadoopOutputFormat(new DummyOutputFormat(),
+			Job.getInstance(), null, null, new Configuration());
+
+		hadoopOutputFormat.finalizeGlobal(1);
+
+		verify(hadoopOutputFormat.outputCommitter, times(1)).commitJob(any(JobContext.class));
+	}
+
+	private OutputCommitter setupOutputCommitter(boolean needsTaskCommit) throws IOException {
+		OutputCommitter outputCommitter = Mockito.mock(OutputCommitter.class);
+		when(outputCommitter.needsTaskCommit(any(TaskAttemptContext.class))).thenReturn(needsTaskCommit);
+		doNothing().when(outputCommitter).commitTask(any(TaskAttemptContext.class));
+
+		return outputCommitter;
+	}
+
+	private HadoopOutputFormat<String, Long> setupHadoopOutputFormat(
+		OutputFormat<String, Long> outputFormat,
+		Job job,
+		RecordWriter<String, Long> recordWriter,
+		OutputCommitter outputCommitter,
+		Configuration configuration) {
+
+		HadoopOutputFormat<String, Long> hadoopOutputFormat = new HadoopOutputFormat<>(outputFormat, job);
+		hadoopOutputFormat.recordWriter = recordWriter;
+		hadoopOutputFormat.outputCommitter = outputCommitter;
+		hadoopOutputFormat.configuration = configuration;
+		hadoopOutputFormat.configuration.set(MAPRED_OUTPUT_DIR_KEY, MAPRED_OUTPUT_PATH);
+
+		return hadoopOutputFormat;
+	}
+
+	class DummyRecordWriter extends RecordWriter<String, Long> {
+		@Override
+		public void write(String key, Long value) throws IOException, InterruptedException {
+		}
+
+		@Override
+		public void close(TaskAttemptContext context) throws IOException, InterruptedException {
+
+		}
+	}
+
+	class DummyOutputFormat extends OutputFormat<String, Long> {
+		@Override
+		public RecordWriter<String, Long> getRecordWriter(TaskAttemptContext context) throws IOException, InterruptedException {
+			return null;
+		}
+
+		@Override
+		public void checkOutputSpecs(JobContext context) throws IOException, InterruptedException {
+
+		}
+
+		@Override
+		public OutputCommitter getOutputCommitter(TaskAttemptContext context) throws IOException, InterruptedException {
+			final OutputCommitter outputCommitter = Mockito.mock(OutputCommitter.class);
+			doNothing().when(outputCommitter).setupJob(any(JobContext.class));
+
+			return outputCommitter;
+		}
+	}
+
+	class ConfigurableDummyOutputFormat extends DummyOutputFormat implements Configurable {
+
+		@Override
+		public void setConf(Configuration configuration) {}
+
+		@Override
+		public Configuration getConf() {
+			return null;
+		}
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/4beff13e/flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/test/hadoopcompatibility/mapred/HadoopIOFormatsITCase.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/test/hadoopcompatibility/mapred/HadoopIOFormatsITCase.java b/flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/test/hadoopcompatibility/mapred/HadoopIOFormatsITCase.java
new file mode 100644
index 0000000..bbe6395
--- /dev/null
+++ b/flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/test/hadoopcompatibility/mapred/HadoopIOFormatsITCase.java
@@ -0,0 +1,234 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.test.hadoopcompatibility.mapred;
+
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.hadoop.mapred.HadoopInputFormat;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.test.util.JavaProgramTestBase;
+import org.apache.flink.test.util.TestBaseUtils;
+import org.apache.flink.util.OperatingSystem;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.SequenceFile;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.SequenceFileInputFormat;
+import org.junit.Assume;
+import org.junit.Before;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.net.URI;
+import java.util.Collection;
+import java.util.LinkedList;
+
+/**
+ * Integraiton tests for Hadoop IO formats.
+ */
+@RunWith(Parameterized.class)
+public class HadoopIOFormatsITCase extends JavaProgramTestBase {
+
+	private static final int NUM_PROGRAMS = 2;
+
+	private int curProgId = config.getInteger("ProgramId", -1);
+	private String[] resultPath;
+	private String[] expectedResult;
+	private String sequenceFileInPath;
+	private String sequenceFileInPathNull;
+
+	public HadoopIOFormatsITCase(Configuration config) {
+		super(config);
+	}
+
+	@Before
+	public void checkOperatingSystem() {
+		// FLINK-5164 - see https://wiki.apache.org/hadoop/WindowsProblems
+		Assume.assumeTrue("This test can't run successfully on Windows.", !OperatingSystem.isWindows());
+	}
+
+	@Override
+	protected void preSubmit() throws Exception {
+		resultPath = new String[] {getTempDirPath("result0"), getTempDirPath("result1") };
+
+		File sequenceFile = createAndRegisterTempFile("seqFile");
+		sequenceFileInPath = sequenceFile.toURI().toString();
+
+		// Create a sequence file
+		org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration();
+		FileSystem fs = FileSystem.get(URI.create(sequenceFile.getAbsolutePath()), conf);
+		Path path = new Path(sequenceFile.getAbsolutePath());
+
+		//  ------------------ Long / Text Key Value pair: ------------
+		int kvCount = 4;
+
+		LongWritable key = new LongWritable();
+		Text value = new Text();
+		SequenceFile.Writer writer = null;
+		try {
+			writer = SequenceFile.createWriter(fs, conf, path, key.getClass(), value.getClass());
+			for (int i = 0; i < kvCount; i++) {
+				if (i == 1) {
+					// write key = 0 a bit more often.
+					for (int a = 0; a < 15; a++) {
+						key.set(i);
+						value.set(i + " - somestring");
+						writer.append(key, value);
+					}
+				}
+				key.set(i);
+				value.set(i + " - somestring");
+				writer.append(key, value);
+			}
+		} finally {
+			IOUtils.closeStream(writer);
+		}
+
+		//  ------------------ Long / Text Key Value pair: ------------
+
+		File sequenceFileNull = createAndRegisterTempFile("seqFileNullKey");
+		sequenceFileInPathNull = sequenceFileNull.toURI().toString();
+		path = new Path(sequenceFileInPathNull);
+
+		LongWritable value1 = new LongWritable();
+		SequenceFile.Writer writer1 = null;
+		try {
+			writer1 = SequenceFile.createWriter(fs, conf, path, NullWritable.class, value1.getClass());
+			for (int i = 0; i < kvCount; i++) {
+				value1.set(i);
+				writer1.append(NullWritable.get(), value1);
+			}
+		} finally {
+			IOUtils.closeStream(writer1);
+		}
+	}
+
+	@Override
+	protected void testProgram() throws Exception {
+		expectedResult = HadoopIOFormatPrograms.runProgram(curProgId, resultPath, sequenceFileInPath, sequenceFileInPathNull);
+	}
+
+	@Override
+	protected void postSubmit() throws Exception {
+		for (int i = 0; i < resultPath.length; i++) {
+			compareResultsByLinesInMemory(expectedResult[i], resultPath[i]);
+		}
+	}
+
+	@Parameters
+	public static Collection<Object[]> getConfigurations() throws FileNotFoundException, IOException {
+
+		LinkedList<Configuration> tConfigs = new LinkedList<Configuration>();
+
+		for (int i = 1; i <= NUM_PROGRAMS; i++) {
+			Configuration config = new Configuration();
+			config.setInteger("ProgramId", i);
+			tConfigs.add(config);
+		}
+
+		return TestBaseUtils.toParameterList(tConfigs);
+	}
+
+	private static class HadoopIOFormatPrograms {
+
+		public static String[] runProgram(int progId, String[] resultPath, String sequenceFileInPath, String sequenceFileInPathNull) throws Exception {
+
+			switch(progId) {
+			case 1: {
+				/**
+				 * Test sequence file, including a key access.
+				 */
+				final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+				SequenceFileInputFormat<LongWritable, Text> sfif = new SequenceFileInputFormat<LongWritable, Text>();
+				JobConf hdconf = new JobConf();
+				SequenceFileInputFormat.addInputPath(hdconf, new Path(sequenceFileInPath));
+				HadoopInputFormat<LongWritable, Text> hif = new HadoopInputFormat<LongWritable, Text>(sfif, LongWritable.class, Text.class, hdconf);
+				DataSet<Tuple2<LongWritable, Text>> ds = env.createInput(hif);
+				DataSet<Tuple2<Long, Text>> sumed = ds.map(new MapFunction<Tuple2<LongWritable, Text>, Tuple2<Long, Text>>() {
+					@Override
+					public Tuple2<Long, Text> map(Tuple2<LongWritable, Text> value) throws Exception {
+						return new Tuple2<Long, Text>(value.f0.get(), value.f1);
+					}
+				}).sum(0);
+				sumed.writeAsText(resultPath[0]);
+				DataSet<String> res = ds.distinct(0).map(new MapFunction<Tuple2<LongWritable, Text>, String>() {
+					@Override
+					public String map(Tuple2<LongWritable, Text> value) throws Exception {
+						return value.f1 + " - " + value.f0.get();
+					}
+				});
+				res.writeAsText(resultPath[1]);
+				env.execute();
+
+				// return expected result
+				return 	new String [] {"(21,3 - somestring)", "0 - somestring - 0\n" +
+						"1 - somestring - 1\n" +
+						"2 - somestring - 2\n" +
+						"3 - somestring - 3\n"};
+
+			}
+			case 2: {
+				final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+				SequenceFileInputFormat<NullWritable, LongWritable> sfif = new SequenceFileInputFormat<NullWritable, LongWritable>();
+				JobConf hdconf = new JobConf();
+				SequenceFileInputFormat.addInputPath(hdconf, new Path(sequenceFileInPathNull));
+				HadoopInputFormat<NullWritable, LongWritable> hif = new HadoopInputFormat<NullWritable, LongWritable>(sfif, NullWritable.class, LongWritable.class, hdconf);
+				DataSet<Tuple2<NullWritable, LongWritable>> ds = env.createInput(hif);
+				DataSet<Tuple2<Void, Long>> res = ds.map(new MapFunction<Tuple2<NullWritable, LongWritable>, Tuple2<Void, Long>>() {
+					@Override
+					public Tuple2<Void, Long> map(Tuple2<NullWritable, LongWritable> value) throws Exception {
+						return new Tuple2<Void, Long>(null, value.f1.get());
+					}
+				});
+				DataSet<Tuple2<Void, Long>> res1 = res.groupBy(1).sum(1);
+				res1.writeAsText(resultPath[1]);
+				res.writeAsText(resultPath[0]);
+				env.execute();
+
+				// return expected result
+				return 	new String [] {"(null,2)\n" +
+						"(null,0)\n" +
+						"(null,1)\n" +
+						"(null,3)",
+						"(null,0)\n" +
+						"(null,1)\n" +
+						"(null,2)\n" +
+						"(null,3)"};
+			}
+			default:
+				throw new IllegalArgumentException("Invalid program id");
+			}
+
+		}
+
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/4beff13e/flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/test/hadoopcompatibility/mapred/WordCountMapredITCase.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/test/hadoopcompatibility/mapred/WordCountMapredITCase.java b/flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/test/hadoopcompatibility/mapred/WordCountMapredITCase.java
new file mode 100644
index 0000000..53927e7
--- /dev/null
+++ b/flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/test/hadoopcompatibility/mapred/WordCountMapredITCase.java
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.test.hadoopcompatibility.mapred;
+
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.hadoop.mapred.HadoopOutputFormat;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.hadoopcompatibility.HadoopInputs;
+import org.apache.flink.test.testdata.WordCountData;
+import org.apache.flink.test.util.JavaProgramTestBase;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.OperatingSystem;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.TextInputFormat;
+import org.apache.hadoop.mapred.TextOutputFormat;
+import org.junit.Assume;
+import org.junit.Before;
+
+import static org.apache.flink.hadoopcompatibility.HadoopInputs.readHadoopFile;
+
+/**
+ * Test WordCount with Hadoop input and output "mapred" (legacy) formats.
+ */
+public class WordCountMapredITCase extends JavaProgramTestBase {
+
+	protected String textPath;
+	protected String resultPath;
+
+	@Before
+	public void checkOperatingSystem() {
+		// FLINK-5164 - see https://wiki.apache.org/hadoop/WindowsProblems
+		Assume.assumeTrue("This test can't run successfully on Windows.", !OperatingSystem.isWindows());
+	}
+
+	@Override
+	protected void preSubmit() throws Exception {
+		textPath = createTempFile("text.txt", WordCountData.TEXT);
+		resultPath = getTempDirPath("result");
+	}
+
+	@Override
+	protected void postSubmit() throws Exception {
+		compareResultsByLinesInMemory(WordCountData.COUNTS, resultPath, new String[] {".", "_"});
+	}
+
+	@Override
+	protected void testProgram() throws Exception {
+		internalRun(true);
+		postSubmit();
+		resultPath = getTempDirPath("result2");
+		internalRun(false);
+		postSubmit();
+	}
+
+	private void internalRun(boolean isTestDeprecatedAPI) throws Exception {
+		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+		DataSet<Tuple2<LongWritable, Text>> input;
+
+		if (isTestDeprecatedAPI) {
+			input = env.createInput(HadoopInputs.readHadoopFile(new TextInputFormat(),
+				LongWritable.class, Text.class, textPath));
+		} else {
+			input = env.createInput(readHadoopFile(new TextInputFormat(),
+				LongWritable.class, Text.class, textPath));
+		}
+
+		DataSet<String> text = input.map(new MapFunction<Tuple2<LongWritable, Text>, String>() {
+			@Override
+			public String map(Tuple2<LongWritable, Text> value) throws Exception {
+				return value.f1.toString();
+			}
+		});
+
+		DataSet<Tuple2<String, Integer>> counts =
+				// split up the lines in pairs (2-tuples) containing: (word,1)
+				text.flatMap(new Tokenizer())
+						// group by the tuple field "0" and sum up tuple field "1"
+						.groupBy(0)
+						.sum(1);
+
+		DataSet<Tuple2<Text, LongWritable>> words = counts.map(new MapFunction<Tuple2<String, Integer>, Tuple2<Text, LongWritable>>() {
+
+			@Override
+			public Tuple2<Text, LongWritable> map(Tuple2<String, Integer> value) throws Exception {
+				return new Tuple2<Text, LongWritable>(new Text(value.f0), new LongWritable(value.f1));
+			}
+		});
+
+		// Set up Hadoop Output Format
+		HadoopOutputFormat<Text, LongWritable> hadoopOutputFormat =
+				new HadoopOutputFormat<Text, LongWritable>(new TextOutputFormat<Text, LongWritable>(), new JobConf());
+		hadoopOutputFormat.getJobConf().set("mapred.textoutputformat.separator", " ");
+		TextOutputFormat.setOutputPath(hadoopOutputFormat.getJobConf(), new Path(resultPath));
+
+		// Output & Execute
+		words.output(hadoopOutputFormat);
+		env.execute("Hadoop Compat WordCount");
+	}
+
+	static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
+
+		@Override
+		public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
+			String[] tokens = value.toLowerCase().split("\\W+");
+			for (String token : tokens) {
+				if (token.length() > 0) {
+					out.collect(new Tuple2<>(token, 1));
+				}
+			}
+		}
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/4beff13e/flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/test/hadoopcompatibility/mapreduce/WordCountMapreduceITCase.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/test/hadoopcompatibility/mapreduce/WordCountMapreduceITCase.java b/flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/test/hadoopcompatibility/mapreduce/WordCountMapreduceITCase.java
new file mode 100644
index 0000000..be70782
--- /dev/null
+++ b/flink-connectors/flink-hadoop-compatibility/src/test/java/org/apache/flink/test/hadoopcompatibility/mapreduce/WordCountMapreduceITCase.java
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.test.hadoopcompatibility.mapreduce;
+
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.hadoop.mapreduce.HadoopOutputFormat;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.hadoopcompatibility.HadoopInputs;
+import org.apache.flink.test.testdata.WordCountData;
+import org.apache.flink.test.util.JavaProgramTestBase;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.OperatingSystem;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
+import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
+import org.junit.Assume;
+import org.junit.Before;
+
+import static org.apache.flink.hadoopcompatibility.HadoopInputs.readHadoopFile;
+
+/**
+ * Test WordCount with Hadoop input and output "mapreduce" (modern) formats.
+ */
+public class WordCountMapreduceITCase extends JavaProgramTestBase {
+
+	protected String textPath;
+	protected String resultPath;
+
+	@Before
+	public void checkOperatingSystem() {
+		// FLINK-5164 - see https://wiki.apache.org/hadoop/WindowsProblems
+		Assume.assumeTrue("This test can't run successfully on Windows.", !OperatingSystem.isWindows());
+	}
+
+	@Override
+	protected void preSubmit() throws Exception {
+		textPath = createTempFile("text.txt", WordCountData.TEXT);
+		resultPath = getTempDirPath("result");
+	}
+
+	@Override
+	protected void postSubmit() throws Exception {
+		compareResultsByLinesInMemory(WordCountData.COUNTS, resultPath, new String[] {".", "_"});
+	}
+
+	@Override
+	protected void testProgram() throws Exception {
+		internalRun(true);
+		postSubmit();
+		resultPath = getTempDirPath("result2");
+		internalRun(false);
+		postSubmit();
+	}
+
+	private void internalRun(boolean isTestDeprecatedAPI) throws Exception {
+		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+		DataSet<Tuple2<LongWritable, Text>> input;
+		if (isTestDeprecatedAPI) {
+			input = env.createInput(HadoopInputs.readHadoopFile(new TextInputFormat(),
+				LongWritable.class, Text.class, textPath));
+		} else {
+			input = env.createInput(readHadoopFile(new TextInputFormat(),
+				LongWritable.class, Text.class, textPath));
+		}
+
+		DataSet<String> text = input.map(new MapFunction<Tuple2<LongWritable, Text>, String>() {
+			@Override
+			public String map(Tuple2<LongWritable, Text> value) throws Exception {
+				return value.f1.toString();
+			}
+		});
+
+		DataSet<Tuple2<String, Integer>> counts =
+				// split up the lines in pairs (2-tuples) containing: (word,1)
+				text.flatMap(new Tokenizer())
+						// group by the tuple field "0" and sum up tuple field "1"
+						.groupBy(0)
+						.sum(1);
+
+		DataSet<Tuple2<Text, LongWritable>> words = counts.map(new MapFunction<Tuple2<String, Integer>, Tuple2<Text, LongWritable>>() {
+
+			@Override
+			public Tuple2<Text, LongWritable> map(Tuple2<String, Integer> value) throws Exception {
+				return new Tuple2<Text, LongWritable>(new Text(value.f0), new LongWritable(value.f1));
+			}
+		});
+
+		// Set up Hadoop Output Format
+		Job job = Job.getInstance();
+		HadoopOutputFormat<Text, LongWritable> hadoopOutputFormat =
+				new HadoopOutputFormat<Text, LongWritable>(new TextOutputFormat<Text, LongWritable>(), job);
+		job.getConfiguration().set("mapred.textoutputformat.separator", " ");
+		TextOutputFormat.setOutputPath(job, new Path(resultPath));
+
+		// Output & Execute
+		words.output(hadoopOutputFormat);
+		env.execute("Hadoop Compat WordCount");
+	}
+
+	static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
+
+		@Override
+		public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
+			String[] tokens = value.toLowerCase().split("\\W+");
+			for (String token : tokens) {
+				if (token.length() > 0) {
+					out.collect(new Tuple2<>(token, 1));
+				}
+			}
+		}
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/4beff13e/flink-connectors/flink-hadoop-compatibility/src/test/scala/org/apache/flink/api/hadoopcompatibility/scala/WordCountMapredITCase.scala
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-hadoop-compatibility/src/test/scala/org/apache/flink/api/hadoopcompatibility/scala/WordCountMapredITCase.scala b/flink-connectors/flink-hadoop-compatibility/src/test/scala/org/apache/flink/api/hadoopcompatibility/scala/WordCountMapredITCase.scala
new file mode 100644
index 0000000..5aaf379
--- /dev/null
+++ b/flink-connectors/flink-hadoop-compatibility/src/test/scala/org/apache/flink/api/hadoopcompatibility/scala/WordCountMapredITCase.scala
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.api.hadoopcompatibility.scala
+
+import org.apache.flink.api.scala._
+import org.apache.flink.api.scala.hadoop.mapred.HadoopOutputFormat
+import org.apache.flink.hadoopcompatibility.scala.HadoopInputs
+import org.apache.flink.test.testdata.WordCountData
+import org.apache.flink.test.util.{JavaProgramTestBase, TestBaseUtils}
+import org.apache.flink.util.OperatingSystem
+import org.apache.hadoop.fs.Path
+import org.apache.hadoop.io.{LongWritable, Text}
+import org.apache.hadoop.mapred.{FileOutputFormat, JobConf, TextInputFormat, TextOutputFormat}
+import org.junit.{Assume, Before}
+
+class WordCountMapredITCase extends JavaProgramTestBase {
+  protected var textPath: String = null
+  protected var resultPath: String = null
+
+  @Before
+  def checkOperatingSystem() {
+    // FLINK-5164 - see https://wiki.apache.org/hadoop/WindowsProblems
+    Assume.assumeTrue("This test can't run successfully on Windows.", !OperatingSystem.isWindows)
+  }
+
+  protected override def preSubmit() {
+    textPath = createTempFile("text.txt", WordCountData.TEXT)
+    resultPath = getTempDirPath("result")
+  }
+
+  protected override def postSubmit() {
+    TestBaseUtils.compareResultsByLinesInMemory(WordCountData.COUNTS,
+                                                resultPath, Array[String](".", "_"))
+  }
+
+  private def internalRun (testDeprecatedAPI: Boolean): Unit = {
+    val env = ExecutionEnvironment.getExecutionEnvironment
+
+    val input =
+      if (testDeprecatedAPI) {
+        env.createInput(
+          HadoopInputs.readHadoopFile(
+            new TextInputFormat, classOf[LongWritable], classOf[Text], textPath))
+      } else {
+        env.createInput(HadoopInputs.readHadoopFile(new TextInputFormat, classOf[LongWritable],
+          classOf[Text], textPath))
+      }
+
+    val counts = input
+      .map(_._2.toString)
+      .flatMap(_.toLowerCase.split("\\W+").filter(_.nonEmpty).map( (_, 1)))
+      .groupBy(0)
+      .sum(1)
+
+    val words = counts
+      .map( t => (new Text(t._1), new LongWritable(t._2)) )
+
+    val hadoopOutputFormat = new HadoopOutputFormat[Text, LongWritable](
+      new TextOutputFormat[Text, LongWritable],
+      new JobConf)
+    hadoopOutputFormat.getJobConf.set("mapred.textoutputformat.separator", " ")
+
+    FileOutputFormat.setOutputPath(hadoopOutputFormat.getJobConf, new Path(resultPath))
+
+    words.output(hadoopOutputFormat)
+
+    env.execute("Hadoop Compat WordCount")
+  }
+
+  protected def testProgram() {
+    internalRun(testDeprecatedAPI = true)
+    postSubmit()
+    resultPath = getTempDirPath("result2")
+    internalRun(testDeprecatedAPI = false)
+    postSubmit()
+  }
+}
+

http://git-wip-us.apache.org/repos/asf/flink/blob/4beff13e/flink-connectors/flink-hadoop-compatibility/src/test/scala/org/apache/flink/api/hadoopcompatibility/scala/WordCountMapreduceITCase.scala
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-hadoop-compatibility/src/test/scala/org/apache/flink/api/hadoopcompatibility/scala/WordCountMapreduceITCase.scala b/flink-connectors/flink-hadoop-compatibility/src/test/scala/org/apache/flink/api/hadoopcompatibility/scala/WordCountMapreduceITCase.scala
new file mode 100644
index 0000000..61e7a12
--- /dev/null
+++ b/flink-connectors/flink-hadoop-compatibility/src/test/scala/org/apache/flink/api/hadoopcompatibility/scala/WordCountMapreduceITCase.scala
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.hadoopcompatibility.scala
+
+import org.apache.flink.api.scala._
+import org.apache.flink.api.scala.hadoop.mapreduce.HadoopOutputFormat
+import org.apache.flink.hadoopcompatibility.scala.HadoopInputs
+import org.apache.flink.test.testdata.WordCountData
+import org.apache.flink.test.util.{JavaProgramTestBase, TestBaseUtils}
+import org.apache.flink.util.OperatingSystem
+import org.apache.hadoop.fs.Path
+import org.apache.hadoop.io.{LongWritable, Text}
+import org.apache.hadoop.mapreduce.Job
+import org.apache.hadoop.mapreduce.lib.input.TextInputFormat
+import org.apache.hadoop.mapreduce.lib.output.{FileOutputFormat, TextOutputFormat}
+import org.junit.{Assume, Before}
+
+class WordCountMapreduceITCase extends JavaProgramTestBase {
+  protected var textPath: String = null
+  protected var resultPath: String = null
+
+  @Before
+  def checkOperatingSystem() {
+    // FLINK-5164 - see https://wiki.apache.org/hadoop/WindowsProblems
+    Assume.assumeTrue("This test can't run successfully on Windows.", !OperatingSystem.isWindows)
+  }
+
+  protected override def preSubmit() {
+    textPath = createTempFile("text.txt", WordCountData.TEXT)
+    resultPath = getTempDirPath("result")
+  }
+
+  protected override def postSubmit() {
+    TestBaseUtils.compareResultsByLinesInMemory(WordCountData.COUNTS,
+                                                resultPath, Array[String](".", "_"))
+  }
+
+  protected def testProgram() {
+    internalRun(testDeprecatedAPI = true)
+    postSubmit()
+    resultPath = getTempDirPath("result2")
+    internalRun(testDeprecatedAPI = false)
+    postSubmit()
+  }
+
+  private def internalRun (testDeprecatedAPI: Boolean): Unit = {
+    val env = ExecutionEnvironment.getExecutionEnvironment
+
+    val input =
+      if (testDeprecatedAPI) {
+        env.createInput(
+          HadoopInputs.readHadoopFile(
+            new TextInputFormat, classOf[LongWritable], classOf[Text], textPath))
+      } else {
+        env.createInput(HadoopInputs.readHadoopFile(new TextInputFormat, classOf[LongWritable],
+          classOf[Text], textPath))
+      }
+
+    val counts = input
+      .map(_._2.toString)
+      .flatMap(_.toLowerCase.split("\\W+").filter(_.nonEmpty).map( (_, 1)))
+      .groupBy(0)
+      .sum(1)
+
+    val words = counts
+      .map( t => (new Text(t._1), new LongWritable(t._2)) )
+
+    val job = Job.getInstance()
+    val hadoopOutputFormat = new HadoopOutputFormat[Text, LongWritable](
+      new TextOutputFormat[Text, LongWritable],
+      job)
+    hadoopOutputFormat.getConfiguration.set("mapred.textoutputformat.separator", " ")
+
+    FileOutputFormat.setOutputPath(job, new Path(resultPath))
+
+    words.output(hadoopOutputFormat)
+
+    env.execute("Hadoop Compat WordCount")
+  }
+}
+

http://git-wip-us.apache.org/repos/asf/flink/blob/4beff13e/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java b/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
index 089c90b..02acfc6 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
@@ -33,7 +33,6 @@ import org.apache.flink.api.common.operators.OperatorInformation;
 import org.apache.flink.api.common.restartstrategy.RestartStrategies;
 import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.hadoop.mapred.HadoopInputFormat;
 import org.apache.flink.api.java.io.CollectionInputFormat;
 import org.apache.flink.api.java.io.CsvReader;
 import org.apache.flink.api.java.io.IteratorInputFormat;
@@ -61,8 +60,6 @@ import org.apache.flink.util.SplittableIterator;
 import org.apache.flink.util.Visitor;
 
 import com.esotericsoftware.kryo.Serializer;
-import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.mapreduce.Job;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -582,109 +579,6 @@ public abstract class ExecutionEnvironment {
 		return new DataSource<>(this, inputFormat, producedType, Utils.getCallLocationName());
 	}
 
-	// ----------------------------------- Hadoop Input Format ---------------------------------------
-
-	/**
-	 * Creates a {@link DataSet} from the given {@link org.apache.hadoop.mapred.FileInputFormat}.
-	 *
-	 * @deprecated Please use {@code org.apache.flink.hadoopcompatibility.HadoopInputs#readHadoopFile(org.apache.hadoop.mapred.FileInputFormat<K,V>, Class<K>, Class<V>, String, JobConf)}
-	 * from the flink-hadoop-compatibility module.
-	 */
-	@Deprecated
-	@PublicEvolving
-	public <K, V> DataSource<Tuple2<K, V>> readHadoopFile(org.apache.hadoop.mapred.FileInputFormat<K, V> mapredInputFormat, Class<K> key, Class<V> value, String inputPath, JobConf job) {
-		DataSource<Tuple2<K, V>> result = createHadoopInput(mapredInputFormat, key, value, job);
-
-		org.apache.hadoop.mapred.FileInputFormat.addInputPath(job, new org.apache.hadoop.fs.Path(inputPath));
-
-		return result;
-	}
-
-	/**
-	 * Creates a {@link DataSet} from {@link org.apache.hadoop.mapred.SequenceFileInputFormat}
-	 * A {@link org.apache.hadoop.mapred.JobConf} with the given inputPath is created.
-	 *
-	 * @deprecated Please use {@code org.apache.flink.hadoopcompatibility.HadoopInputs#readSequenceFile(Class<K>, Class<V>, String)}
-	 * from the flink-hadoop-compatibility module.
-	 */
-	@Deprecated
-	@PublicEvolving
-	public <K, V> DataSource<Tuple2<K, V>> readSequenceFile(Class<K> key, Class<V> value, String inputPath) throws IOException {
-		return readHadoopFile(new org.apache.hadoop.mapred.SequenceFileInputFormat<K, V>(), key, value, inputPath);
-	}
-
-	/**
-	 * Creates a {@link DataSet} from the given {@link org.apache.hadoop.mapred.FileInputFormat}. A
-	 * {@link org.apache.hadoop.mapred.JobConf} with the given inputPath is created.
-	 *
-	 * @deprecated Please use {@code org.apache.flink.hadoopcompatibility.HadoopInputs#readHadoopFile(org.apache.hadoop.mapred.FileInputFormat<K,V>, Class<K>, Class<V>, String)}
-	 * from the flink-hadoop-compatibility module.
-	 */
-	@Deprecated
-	@PublicEvolving
-	public <K, V> DataSource<Tuple2<K, V>> readHadoopFile(org.apache.hadoop.mapred.FileInputFormat<K, V> mapredInputFormat, Class<K> key, Class<V> value, String inputPath) {
-		return readHadoopFile(mapredInputFormat, key, value, inputPath, new JobConf());
-	}
-
-	/**
-	 * Creates a {@link DataSet} from the given {@link org.apache.hadoop.mapred.InputFormat}.
-	 *
-	 * @deprecated Please use {@code org.apache.flink.hadoopcompatibility.HadoopInputs#createHadoopInput(org.apache.hadoop.mapred.InputFormat<K,V>, Class<K>, Class<V>, JobConf)}
-	 * from the flink-hadoop-compatibility module.
-	 */
-	@Deprecated
-	@PublicEvolving
-	public <K, V> DataSource<Tuple2<K, V>> createHadoopInput(org.apache.hadoop.mapred.InputFormat<K, V> mapredInputFormat, Class<K> key, Class<V> value, JobConf job) {
-		HadoopInputFormat<K, V> hadoopInputFormat = new HadoopInputFormat<>(mapredInputFormat, key, value, job);
-
-		return this.createInput(hadoopInputFormat);
-	}
-
-	/**
-	 * Creates a {@link DataSet} from the given {@link org.apache.hadoop.mapreduce.lib.input.FileInputFormat}. The
-	 * given inputName is set on the given job.
-	 *
-	 * @deprecated Please use {@code org.apache.flink.hadoopcompatibility.HadoopInputs#readHadoopFile(org.apache.hadoop.mapreduce.lib.input.FileInputFormat<K,V>, Class<K>, Class<V>, String, Job)}
-	 * from the flink-hadoop-compatibility module.
-	 */
-	@Deprecated
-	@PublicEvolving
-	public <K, V> DataSource<Tuple2<K, V>> readHadoopFile(org.apache.hadoop.mapreduce.lib.input.FileInputFormat<K, V> mapreduceInputFormat, Class<K> key, Class<V> value, String inputPath, Job job) throws IOException {
-		DataSource<Tuple2<K, V>> result = createHadoopInput(mapreduceInputFormat, key, value, job);
-
-		org.apache.hadoop.mapreduce.lib.input.FileInputFormat.addInputPath(job, new org.apache
-				.hadoop.fs.Path(inputPath));
-
-		return result;
-	}
-
-	/**
-	 * Creates a {@link DataSet} from the given {@link org.apache.hadoop.mapreduce.lib.input.FileInputFormat}. A
-	 * {@link org.apache.hadoop.mapreduce.Job} with the given inputPath is created.
-	 *
-	 * @deprecated Please use {@code  org.apache.flink.hadoopcompatibility.HadoopInputs#readHadoopFile(org.apache.hadoop.mapreduce.lib.input.FileInputFormat<K,V>, Class<K>, Class<V>, String)}
-	 * from the flink-hadoop-compatibility module.
-	 */
-	@Deprecated
-	@PublicEvolving
-	public <K, V> DataSource<Tuple2<K, V>> readHadoopFile(org.apache.hadoop.mapreduce.lib.input.FileInputFormat<K, V> mapreduceInputFormat, Class<K> key, Class<V> value, String inputPath) throws IOException {
-		return readHadoopFile(mapreduceInputFormat, key, value, inputPath, Job.getInstance());
-	}
-
-	/**
-	 * Creates a {@link DataSet} from the given {@link org.apache.hadoop.mapreduce.InputFormat}.
-	 *
-	 * @deprecated Please use {@code org.apache.flink.hadoopcompatibility.HadoopInputs#createHadoopInput(org.apache.hadoop.mapreduce.InputFormat<K,V>, Class<K>, Class<V>, Job)}
-	 * from the flink-hadoop-compatibility module.
-	 */
-	@Deprecated
-	@PublicEvolving
-	public <K, V> DataSource<Tuple2<K, V>> createHadoopInput(org.apache.hadoop.mapreduce.InputFormat<K, V> mapreduceInputFormat, Class<K> key, Class<V> value, Job job) {
-		org.apache.flink.api.java.hadoop.mapreduce.HadoopInputFormat<K, V> hadoopInputFormat = new org.apache.flink.api.java.hadoop.mapreduce.HadoopInputFormat<>(mapreduceInputFormat, key, value, job);
-
-		return this.createInput(hadoopInputFormat);
-	}
-
 	// ----------------------------------- Collection ---------------------------------------
 
 	/**

http://git-wip-us.apache.org/repos/asf/flink/blob/4beff13e/flink-java/src/main/java/org/apache/flink/api/java/hadoop/common/HadoopInputFormatCommonBase.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/common/HadoopInputFormatCommonBase.java b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/common/HadoopInputFormatCommonBase.java
deleted file mode 100644
index bee331c..0000000
--- a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/common/HadoopInputFormatCommonBase.java
+++ /dev/null
@@ -1,81 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.api.java.hadoop.common;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.io.RichInputFormat;
-import org.apache.flink.core.io.InputSplit;
-
-import org.apache.hadoop.security.Credentials;
-import org.apache.hadoop.security.UserGroupInformation;
-
-import java.io.IOException;
-import java.io.ObjectInputStream;
-import java.io.ObjectOutputStream;
-import java.lang.reflect.InvocationTargetException;
-import java.lang.reflect.Method;
-
-/**
- * A common base for both "mapred" and "mapreduce" Hadoop input formats.
- */
-@Internal
-public abstract class HadoopInputFormatCommonBase<T, SPITTYPE extends InputSplit> extends RichInputFormat<T, SPITTYPE> {
-	protected transient Credentials credentials;
-
-	protected HadoopInputFormatCommonBase(Credentials creds) {
-		this.credentials = creds;
-	}
-
-	protected void write(ObjectOutputStream out) throws IOException {
-		this.credentials.write(out);
-	}
-
-	public void read(ObjectInputStream in) throws IOException {
-		this.credentials = new Credentials();
-		credentials.readFields(in);
-	}
-
-	/**
-	 * This method only exists because there is no UserGroupInformation.getCredentials() method
-	 * in Hadoop 1.x
-	 *
-	 * <p>Note that this method returns "null" in Hadoop 1.x environments.
-	 *
-	 * @param ugi The user information
-	 * @return new credentials object from the user information. MAY RETURN NULL!
-	 */
-	public static Credentials getCredentialsFromUGI(UserGroupInformation ugi) {
-		Method getCredentialsMethod = null;
-		for (Method m : ugi.getClass().getMethods()) {
-			if (m.getName().equals("getCredentials")) {
-				getCredentialsMethod = m;
-				break;
-			}
-		}
-		if (getCredentialsMethod == null) {
-			return null;
-		} else {
-			try {
-				return (Credentials) getCredentialsMethod.invoke(ugi);
-			} catch (InvocationTargetException | IllegalAccessException e) {
-				throw new RuntimeException("Unable to get credentials from UserGroupInformation. This is only supported by Hadoop 2.2.0+");
-			}
-		}
-	}
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/4beff13e/flink-java/src/main/java/org/apache/flink/api/java/hadoop/common/HadoopOutputFormatCommonBase.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/common/HadoopOutputFormatCommonBase.java b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/common/HadoopOutputFormatCommonBase.java
deleted file mode 100644
index 8f45612..0000000
--- a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/common/HadoopOutputFormatCommonBase.java
+++ /dev/null
@@ -1,51 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.api.java.hadoop.common;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.io.RichOutputFormat;
-
-import org.apache.hadoop.security.Credentials;
-
-import java.io.IOException;
-import java.io.ObjectInputStream;
-import java.io.ObjectOutputStream;
-
-/**
- * A common base for both "mapred" and "mapreduce" Hadoop output formats.
- *
- * <p>The base is taking care of handling (serializing) security credentials.
- */
-@Internal
-public abstract class HadoopOutputFormatCommonBase<T> extends RichOutputFormat<T> {
-	protected transient Credentials credentials;
-
-	protected HadoopOutputFormatCommonBase(Credentials creds) {
-		this.credentials = creds;
-	}
-
-	protected void write(ObjectOutputStream out) throws IOException {
-		this.credentials.write(out);
-	}
-
-	public void read(ObjectInputStream in) throws IOException {
-		this.credentials = new Credentials();
-		credentials.readFields(in);
-	}
-}


Mime
View raw message