[FLINK-6294] Fix potential NPE in BucketingSink.close()

d7911c5a · Aljoscha Krettek · d7c2c417 · d7911c5a
隐藏空白更改
内联并排

Showing with 4 addition and 2 deletion

flink-connectors/flink-connector-filesystem/src/main/java/org/apache/flink/streaming/connectors/fs/bucketing/BucketingSink.java ...link/streaming/connectors/fs/bucketing/BucketingSink.java +4 -2

未找到文件。
--- a/flink-connectors/flink-connector-filesystem/src/main/java/org/apache/flink/streaming/connectors/fs/bucketing/BucketingSink.java
+++ b/flink-connectors/flink-connector-filesystem/src/main/java/org/apache/flink/streaming/connectors/fs/bucketing/BucketingSink.java
@@ -414,8 +414,10 @@ public class BucketingSink<T>

 	@Override
 	public void close() throws Exception {
-		for (Map.Entry<String, BucketState<T>> entry : state.bucketStates.entrySet()) {
-			closeCurrentPartFile(entry.getValue());
+		if (state != null) {
+			for (Map.Entry<String, BucketState<T>> entry : state.bucketStates.entrySet()) {
+				closeCurrentPartFile(entry.getValue());
+			}
 		}
 	}