[hotfix] [runtime] Fix processing unmatched build-side when all bucke…

…ts spilled
JacksonSha · Feb 23, 2016 · 03923c3 · 03923c3
1 parent 8656ed5
commit 03923c3
Showing 1 changed file with 11 additions and 1 deletion.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/MutableHashTable.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/MutableHashTable.java
@@ -1810,6 +1810,11 @@ private BT nextInBucket(BT reuse) {
 					}
 				}
 
+				// if all buckets were spilled out of memory
+				if (this.bucketSegment == null) {
+					return null;
+				}
+
 				// this segment is done. check if there is another chained bucket
 				final long forwardPointer = this.bucketSegment.getLong(this.bucketInSegmentOffset + HEADER_FORWARD_OFFSET);
 				if (forwardPointer == BUCKET_FORWARD_POINTER_NOT_SET) {
@@ -1847,7 +1852,12 @@ private BT nextInBucket() {
 						this.numInSegment++;
 					}
 				}
-
+
+				// if all buckets were spilled out of memory
+				if (this.bucketSegment == null) {
+					return null;
+				}
+
 				// this segment is done. check if there is another chained bucket
 				final long forwardPointer = this.bucketSegment.getLong(this.bucketInSegmentOffset + HEADER_FORWARD_OFFSET);
 				if (forwardPointer == BUCKET_FORWARD_POINTER_NOT_SET) {