code: 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);
  			}
  		}
  	}
docs: Repartition BROADCAST state.