public < T > BroadcastVariableMaterialization < T , ? > materializeBroadcastVariable ( String name , int superstep , BatchTask < ? , ? > holder , MutableReader < ? > reader , TypeSerializerFactory < T > serializerFactory ) throws IOException { final BroadcastVariableKey key = new BroadcastVariableKey ( holder . getEnvironment ( ) . getJobVertexId ( ) , name , superstep ) ; while ( true ) { final BroadcastVariableMaterialization < T , Object > newMat = new BroadcastVariableMaterialization < T , Object > ( key ) ; final BroadcastVariableMaterialization < ? , ? > previous = variables . putIfAbsent ( key , newMat ) ; @ SuppressWarnings ( "unchecked" ) final BroadcastVariableMaterialization < T , ? > materialization = ( previous == null ) ? newMat : ( BroadcastVariableMaterialization < T , ? > ) previous ; try { materialization . materializeVariable ( reader , serializerFactory , holder ) ; return materialization ; } catch ( MaterializationExpiredException e ) { // concurrent release. as an optimization, try to replace the previous one with our version. otherwise we might spin for a while // until the releaser removes the variable // NOTE: This would also catch a bug prevented an expired materialization from ever being removed, so it acts as a future safeguard boolean replaceSuccessful = false ; try { replaceSuccessful = variables . replace ( key , materialization , newMat ) ; } catch ( Throwable t ) { } if ( replaceSuccessful ) { try { newMat . materializeVariable ( reader , serializerFactory , holder ) ; return newMat ; } catch ( MaterializationExpiredException ee ) { // can still happen in cases of extreme races and fast tasks // fall through the loop; } } // else fall through the loop } } }
Materializes the broadcast variable for the given name scoped to the given task and its iteration superstep . An existing materialization created by another parallel subtask may be returned if it hasn t expired yet .