@ SuppressWarnings ( "unchecked" ) @ Override public Future < ? > runBatchAsync ( Batch < T > batch , AfterExecute < T > afterExecute ) { for ( Map . Entry < Address , T > entry : batch . getMap ( ) . entrySet ( ) ) { Address address = entry . getKey ( ) ; T opFromBatch = entry . getValue ( ) ; BoxedByteArray mapKey ; try { mapKey = new BoxedByteArray ( address . toWriteKey ( idService ) ) ; } catch ( Exception e ) { throw new RuntimeException ( e ) ; } if ( commitType == CommitType . READ_COMBINE_CAS ) { int casRetriesRemaining = casRetries ; do { Optional < byte [ ] > oldBytes ; try { oldBytes = getRaw ( address ) ; } catch ( Exception e ) { throw new RuntimeException ( e ) ; } T newOp ; if ( oldBytes . isPresent ( ) ) { T oldOp = deserializer . fromBytes ( oldBytes . get ( ) ) ; newOp = ( T ) opFromBatch . add ( oldOp ) ; if ( log . isDebugEnabled ( ) ) { log . debug ( "Combined " + oldOp + " and " + opFromBatch + " into " + newOp ) ; } } else { newOp = opFromBatch ; } if ( oldBytes . isPresent ( ) ) { // Compare and swap, if the key is still mapped to the same byte array. if ( map . replace ( mapKey , oldBytes . get ( ) , newOp . serialize ( ) ) ) { if ( log . isDebugEnabled ( ) ) { log . debug ( "Successful CAS overwrite at key " + Hex . encodeHexString ( mapKey . bytes ) + " with " + Hex . encodeHexString ( newOp . serialize ( ) ) ) ; } break ; } } else { // Compare and swap, if the key is still absent from the map. if ( map . putIfAbsent ( mapKey , newOp . serialize ( ) ) == null ) { // null is returned when there was no existing mapping for the // given key, which is the success case. if ( log . isDebugEnabled ( ) ) { log . debug ( "Successful CAS insert without existing key for key " + Hex . encodeHexString ( mapKey . bytes ) + " with " + Hex . encodeHexString ( newOp . serialize ( ) ) ) ; } break ; } } } while ( casRetriesRemaining -- > 0 ) ; if ( casRetriesRemaining == - 1 ) { RuntimeException e = new RuntimeException ( "CAS retries exhausted" ) ; afterExecute . afterExecute ( e ) ; throw e ; } } else if ( commitType == CommitType . OVERWRITE ) { map . put ( mapKey , opFromBatch . serialize ( ) ) ; if ( log . isDebugEnabled ( ) ) { log . debug ( "Write of key " + Hex . encodeHexString ( mapKey . bytes ) ) ; } } else { throw new AssertionError ( "Unsupported commit type: " + commitType ) ; } } batch . reset ( ) ; afterExecute . afterExecute ( null ) ; // null throwable => success return nullFuture ; }
Actually synchronous and not asyncronous which is allowed .