[FLINK-2826] [runtime] Fix race condition and locking in...

[FLINK-2826] [runtime] Fix race condition and locking in BroadcastVariableMaterialization#decrementReferenceInternal

This closes #1339

flink-runtime/src/main/java/org/apache/flink/runtime/broadcast/BroadcastVariableMaterialization.java

@@ -137,7 +137,7 @@ public class BroadcastVariableMaterialization<T, C> {
 				while ((element = readerIterator.next(element)) != null);
 				
 				synchronized (materializationMonitor) {
-					while (!this.materialized) {
+					while (!this.materialized && !disposed) {
 						materializationMonitor.wait();
 					}
 				}
@@ -209,7 +209,7 @@ public class BroadcastVariableMaterialization<T, C> {
 			throw new IllegalStateException("The Broadcast Variable has been disposed");
 		}
 		
-		synchronized (this) {
+		synchronized (references) {
 			if (transformed != null) {
 				if (transformed instanceof List) {
 					@SuppressWarnings("unchecked")
@@ -233,7 +233,7 @@ public class BroadcastVariableMaterialization<T, C> {
 			throw new IllegalStateException("The Broadcast Variable has been disposed");
 		}
 		
-		synchronized (this) {
+		synchronized (references) {
 			if (transformed == null) {
 				transformed = initializer.initializeBroadcastVariable(data);
 				data = null;