flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From fhue...@apache.org
Subject flink git commit: [FLINK-5518] [hadoopCompat] Add null check to HadoopInputFormatBase.close().
Date Tue, 17 Jan 2017 19:59:47 GMT
Repository: flink
Updated Branches:
  refs/heads/release-1.1 4ea52d69d -> 214c188d7


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


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/214c188d
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/214c188d
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/214c188d

Branch: refs/heads/release-1.1
Commit: 214c188d7fffd66a43325eff392f12e8f3421fab
Parents: 4ea52d6
Author: Jakub Havlik <jakub.havlik@brain-station.com>
Authored: Tue Jan 17 08:26:07 2017 +0100
Committer: Fabian Hueske <fhueske@apache.org>
Committed: Tue Jan 17 20:59:26 2017 +0100

----------------------------------------------------------------------
 .../flink/api/java/hadoop/mapred/HadoopInputFormatBase.java  | 8 +++++---
 .../api/java/hadoop/mapreduce/HadoopInputFormatBase.java     | 8 +++++---
 .../flink/api/java/hadoop/mapred/HadoopInputFormatTest.java  | 7 +++++++
 .../api/java/hadoop/mapreduce/HadoopInputFormatTest.java     | 6 ++++++
 4 files changed, 23 insertions(+), 6 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/214c188d/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormatBase.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormatBase.java
b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormatBase.java
index ef9999f..ae23a49 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormatBase.java
+++ b/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();
+			}
 		}
 	}
 	

http://git-wip-us.apache.org/repos/asf/flink/blob/214c188d/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormatBase.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormatBase.java
b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormatBase.java
index 0335c23..9d8a8c5 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormatBase.java
+++ b/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();
+			}
 		}
 	}
 

http://git-wip-us.apache.org/repos/asf/flink/blob/214c188d/flink-java/src/test/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormatTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormatTest.java
b/flink-java/src/test/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormatTest.java
index 3b8d227..434ad15 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormatTest.java
+++ b/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());
 	}

http://git-wip-us.apache.org/repos/asf/flink/blob/214c188d/flink-java/src/test/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormatTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormatTest.java
b/flink-java/src/test/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormatTest.java
index d6ec484..4c9c009 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormatTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormatTest.java
@@ -99,6 +99,12 @@ public class HadoopInputFormatTest {
 	}
 
 	@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();
 


Mime
View raw message