[FLINK-5518] [hadoopCompat] Add null check to HadoopInputFormatBase.close().

This closes #3133
This closes #243 // closing stale PR

flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormatBase.java

@@ -190,10 +190,12 @@ public abstract class HadoopInputFormatBase<K, V, T> extends HadoopInputFormatCo
 
 	@Override
 	public void close() throws IOException {
+		if (this.recordReader != null) {
 
-		// enforce sequential close() calls
-		synchronized (CLOSE_MUTEX) {
-			this.recordReader.close();
+			// enforce sequential close() calls
+			synchronized (CLOSE_MUTEX) {
+				this.recordReader.close();
+			}
 		}
 	}
 	

flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormatBase.java

@@ -225,10 +225,12 @@ public abstract class HadoopInputFormatBase<K, V, T> extends HadoopInputFormatCo
 
 	@Override
 	public void close() throws IOException {
+		if (this.recordReader != null) {
 
-		// enforce sequential close() calls
-		synchronized (CLOSE_MUTEX) {
-			this.recordReader.close();
+			// enforce sequential close() calls
+			synchronized (CLOSE_MUTEX) {
+				this.recordReader.close();
+			}
 		}
 	}
 

flink-java/src/test/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormatTest.java

@@ -174,6 +174,13 @@ public class HadoopInputFormatTest {
 		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());
 	}

flink-java/src/test/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormatTest.java

@@ -98,6 +98,12 @@ public class HadoopInputFormatTest {
 		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();