private List < Map < StreamStateHandle , OperatorStateHandle > > repartition ( GroupByStateNameResults nameToStateByMode , int newParallelism ) { // We will use this to merge w.r.t. StreamStateHandles for each parallel subtask inside the maps List < Map < StreamStateHandle , OperatorStateHandle > > mergeMapList = new ArrayList <> ( newParallelism ) ; // Initialize for ( int i = 0 ; i < newParallelism ; ++ i ) { mergeMapList . add ( new HashMap <> ( ) ) ; } // Start with the state handles we distribute round robin by splitting by offsets Map < String , List < Tuple2 < StreamStateHandle , OperatorStateHandle . StateMetaInfo > > > nameToDistributeState = nameToStateByMode . getByMode ( OperatorStateHandle . Mode . SPLIT_DISTRIBUTE ) ; repartitionSplitState ( nameToDistributeState , newParallelism , mergeMapList ) ; // Now we also add the state handles marked for union to all parallel instances Map < String , List < Tuple2 < StreamStateHandle , OperatorStateHandle . StateMetaInfo > > > nameToUnionState = nameToStateByMode . getByMode ( OperatorStateHandle . Mode . UNION ) ; repartitionUnionState ( nameToUnionState , mergeMapList ) ; // Now we also add the state handles marked for uniform broadcast to all parallel instances Map < String , List < Tuple2 < StreamStateHandle , OperatorStateHandle . StateMetaInfo > > > nameToBroadcastState = nameToStateByMode . getByMode ( OperatorStateHandle . Mode . BROADCAST ) ; repartitionBroadcastState ( nameToBroadcastState , mergeMapList ) ; return mergeMapList ; }
Repartition all named states .