@ SuppressWarnings ( "unchecked, rawtype" ) private GroupByStateNameResults groupByStateMode ( List < List < OperatorStateHandle > > previousParallelSubtaskStates ) { //Reorganize: group by (State Name -> StreamStateHandle + StateMetaInfo) EnumMap < OperatorStateHandle . Mode , Map < String , List < Tuple2 < StreamStateHandle , OperatorStateHandle . StateMetaInfo > > > > nameToStateByMode = new EnumMap <> ( OperatorStateHandle . Mode . class ) ; for ( OperatorStateHandle . Mode mode : OperatorStateHandle . Mode . values ( ) ) { nameToStateByMode . put ( mode , new HashMap <> ( ) ) ; } for ( List < OperatorStateHandle > previousParallelSubtaskState : previousParallelSubtaskStates ) { for ( OperatorStateHandle operatorStateHandle : previousParallelSubtaskState ) { if ( operatorStateHandle == null ) { continue ; } final Set < Map . Entry < String , OperatorStateHandle . StateMetaInfo > > partitionOffsetEntries = operatorStateHandle . getStateNameToPartitionOffsets ( ) . entrySet ( ) ; for ( Map . Entry < String , OperatorStateHandle . StateMetaInfo > e : partitionOffsetEntries ) { OperatorStateHandle . StateMetaInfo metaInfo = e . getValue ( ) ; Map < String , List < Tuple2 < StreamStateHandle , OperatorStateHandle . StateMetaInfo > > > nameToState = nameToStateByMode . get ( metaInfo . getDistributionMode ( ) ) ; List < Tuple2 < StreamStateHandle , OperatorStateHandle . StateMetaInfo > > stateLocations = nameToState . computeIfAbsent ( e . getKey ( ) , k -> new ArrayList <> ( previousParallelSubtaskStates . size ( ) * partitionOffsetEntries . size ( ) ) ) ; stateLocations . add ( Tuple2 . of ( operatorStateHandle . getDelegateStateHandle ( ) , e . getValue ( ) ) ) ; } } } return new GroupByStateNameResults ( nameToStateByMode ) ; }
Group by the different named states .