You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fh...@apache.org on 2017/01/17 19:55:29 UTC
flink git commit: [FLINK-5518] [hadoopCompat] Add null check to
HadoopInputFormatBase.close().
Repository: flink
Updated Branches:
refs/heads/master e187b5ee9 -> ece899a9f
[FLINK-5518] [hadoopCompat] Add null check to HadoopInputFormatBase.close().
This closes #3133
This closes #243 // closing stale PR
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/ece899a9
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/ece899a9
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/ece899a9
Branch: refs/heads/master
Commit: ece899a9f33b3fe4aa35d6cc66c43053170e631f
Parents: e187b5e
Author: Jakub Havlik <ja...@brain-station.com>
Authored: Tue Jan 17 08:26:07 2017 +0100
Committer: Fabian Hueske <fh...@apache.org>
Committed: Tue Jan 17 16:50:15 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/ece899a9/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/ece899a9/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/ece899a9/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/ece899a9/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();