docs: public void materializeVariable(MutableReader<?> reader, TypeSerializerFactory<?> serializerFactory, BatchTask<?, ?> referenceHolder)
  			throws MaterializationExpiredException, IOException {
  		Preconditions.checkNotNull(reader);
  		Preconditions.checkNotNull(serializerFactory);
  		Preconditions.checkNotNull(referenceHolder);
  
  		final boolean materializer;
  
  		// hold the reference lock only while we track references and decide who should be the materializer
  		// that way, other tasks can de-register (in case of failure) while materialization is happening
  		synchronized (references) {
  			if (disposed) {
  				throw new MaterializationExpiredException();
  			}
  
  			// sanity check
  			if (!references.add(referenceHolder)) {
  				throw new IllegalStateException(
  						String.format("The task %s already holds a reference to the broadcast variable %s.",
  								referenceHolder.getEnvironment().getTaskInfo().getTaskNameWithSubtasks(),
  								key.toString()));
  			}
  
  			materializer = references.size() == 1;
  		}
  
  		try {
  			@SuppressWarnings("unchecked")
  			final MutableReader<DeserializationDelegate<T>> typedReader = (MutableReader<DeserializationDelegate<T>>) reader;
  
  			@SuppressWarnings("unchecked")
  			final TypeSerializer<T> serializer = ((TypeSerializerFactory<T>) serializerFactory).getSerializer();
  
  			final ReaderIterator<T> readerIterator = new ReaderIterator<T>(typedReader, serializer);
  
  			if (materializer) {
  				// first one, so we need to materialize;
  				if (LOG.isDebugEnabled()) {
  					LOG.debug("Getting Broadcast Variable (" + key + ") - First access, materializing.");
  				}
  
  				ArrayList<T> data = new ArrayList<T>();
  
  				T element;
  				while ((element = readerIterator.next()) != null) {
  					data.add(element);
  				}
  
  				synchronized (materializationMonitor) {
  					this.data = data;
  					this.materialized = true;
  					materializationMonitor.notifyAll();
  				}
  
  				if (LOG.isDebugEnabled()) {
  					LOG.debug("Materialization of Broadcast Variable (" + key + ") finished.");
  				}
  			}
  			else {
  				// successor: discard all data and refer to the shared variable
  
  				if (LOG.isDebugEnabled()) {
  					LOG.debug("Getting Broadcast Variable (" + key + ") - shared access.");
  				}
  
  				T element = serializer.createInstance();
  				while ((element = readerIterator.next(element)) != null) {
  				}
  
  				synchronized (materializationMonitor) {
  					while (!this.materialized && !disposed) {
  						materializationMonitor.wait();
  					}
  				}
  
  			}
  		}
  		catch (Throwable t) {
  			// in case of an exception, we need to clean up big time
  			decrementReferenceIfHeld(referenceHolder);
  
  			if (t instanceof IOException) {
  				throw (IOException) t;
  			} else {
  				throw new IOException("Materialization of the broadcast variable failed.", t);
  			}
  		}
  	}