private void sendKeysInBatches ( MapStoreContext mapStoreContext , boolean replaceExistingValues ) throws Exception { if ( logger . isFinestEnabled ( ) ) { logger . finest ( "sendKeysInBatches invoked " + getStateMessage ( ) ) ; } int clusterSize = partitionService . getMemberPartitionsMap ( ) . size ( ) ; Iterator < Object > keys = null ; Throwable loadError = null ; try { Iterable < Object > allKeys = mapStoreContext . loadAllKeys ( ) ; keys = allKeys . iterator ( ) ; Iterator < Data > dataKeys = map ( keys , toData ) ; int mapMaxSize = clusterSize * maxSizePerNode ; if ( mapMaxSize > 0 ) { dataKeys = limit ( dataKeys , mapMaxSize ) ; } Iterator < Entry < Integer , Data > > partitionsAndKeys = map ( dataKeys , toPartition ( partitionService ) ) ; Iterator < Map < Integer , List < Data > > > batches = toBatches ( partitionsAndKeys , maxBatch ) ; List < Future > futures = new ArrayList <> ( ) ; while ( batches . hasNext ( ) ) { Map < Integer , List < Data > > batch = batches . next ( ) ; futures . addAll ( sendBatch ( batch , replaceExistingValues ) ) ; } // This acts as a barrier to prevent re-ordering of key distribution operations (LoadAllOperation) // and LoadStatusOperation(s) which indicates all keys were already loaded. // Re-ordering of in-flight operations can happen during a partition migration. We are waiting here // for all LoadAllOperation(s) to be ACKed by receivers and only then we send them the LoadStatusOperation // See https://github.com/hazelcast/hazelcast/issues/4024 for additional details FutureUtil . waitForever ( futures ) ; } catch ( Exception caught ) { loadError = caught ; } finally { sendKeyLoadCompleted ( clusterSize , loadError ) ; if ( keys instanceof Closeable ) { closeResource ( ( Closeable ) keys ) ; } } }
Loads keys from the map loader and sends them to the partition owners in batches for value loading . This method will return after all keys have been dispatched to the partition owners for value loading and all partitions have been notified that the key loading has completed . The values will still be loaded asynchronously and can be put into the record stores after this method has returned . If there is a configured max size policy per node the keys will be loaded until this many keys have been loaded from the map loader . If the keys returned from the map loader are not equally distributed over all partitions this may cause some nodes to load more entries than others and exceed the configured policy .