@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;
    }
@ 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 ( ) ) { 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 { if ( map . putIfAbsent ( mapKey , newOp . serialize ( ) ) == null ) { 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 ) ; return nullFuture ; }
Actually synchronous and not asyncronous, which is allowed.