private void repartitionBroadcastState ( Map < String , List < Tuple2 < StreamStateHandle , OperatorStateHandle . StateMetaInfo > > > broadcastState , List < Map < StreamStateHandle , OperatorStateHandle > > mergeMapList ) { int newParallelism = mergeMapList . size ( ) ; for ( int i = 0 ; i < newParallelism ; ++ i ) { final Map < StreamStateHandle , OperatorStateHandle > mergeMap = mergeMapList . get ( i ) ; // for each name, pick the i-th entry for ( Map . Entry < String , List < Tuple2 < StreamStateHandle , OperatorStateHandle . StateMetaInfo > > > e : broadcastState . entrySet ( ) ) { int previousParallelism = e . getValue ( ) . size ( ) ; Tuple2 < StreamStateHandle , OperatorStateHandle . StateMetaInfo > handleWithMetaInfo = e . getValue ( ) . get ( i % previousParallelism ) ; OperatorStateHandle operatorStateHandle = mergeMap . get ( handleWithMetaInfo . f0 ) ; if ( operatorStateHandle == null ) { operatorStateHandle = new OperatorStreamStateHandle ( new HashMap <> ( broadcastState . size ( ) ) , handleWithMetaInfo . f0 ) ; mergeMap . put ( handleWithMetaInfo . f0 , operatorStateHandle ) ; } operatorStateHandle . getStateNameToPartitionOffsets ( ) . put ( e . getKey ( ) , handleWithMetaInfo . f1 ) ; } } }
Repartition BROADCAST state .