[FLINK-17080][java,table] Fix possible NPE in Utils#CollectHelper

This closes apache#11688
shuttie · Apr 15, 2020 · 0fa0830 · 0fa0830
1 parent ffb9878
commit 0fa0830
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 3 deletions.
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/Utils.java b/flink-java/src/main/java/org/apache/flink/api/java/Utils.java
@@ -138,8 +138,13 @@ public void writeRecord(T record) throws IOException {
 
 		@Override
 		public void close() {
-			// Important: should only be added in close method to minimize traffic of accumulators
-			getRuntimeContext().addAccumulator(id, accumulator);
+			// when the sink is up but not initialized and the job fails due to other operators,
+			// it is possible that close() is called when open() is not called,
+			// so we have to do this null check
+			if (accumulator != null) {
+				// Important: should only be added in close method to minimize traffic of accumulators
+				getRuntimeContext().addAccumulator(id, accumulator);
+			}
 		}
 	}
 

diff --git a/...-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/api/TableUtils.java b/...-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/api/TableUtils.java
@@ -84,7 +84,11 @@ public static List<Row> collectToList(Table table) throws Exception {
 			tEnv.insertInto(sinkName, table);
 			JobExecutionResult executionResult = tEnv.execute(jobName);
 			ArrayList<byte[]> accResult = executionResult.getAccumulatorResult(id);
-			deserializedList = SerializedListAccumulator.deserializeList(accResult, serializer);
+			if (accResult != null) {
+				deserializedList = SerializedListAccumulator.deserializeList(accResult, serializer);
+			} else {
+				throw new RuntimeException("Could not retrieve table result. It is very likely that the job fails.");
+			}
 		} finally {
 			tEnv.dropTemporaryTable(sinkName);
 		}