CompletableFuture < List < Long > > update ( DirectSegmentAccess segment , TableKeyBatch batch , Supplier < CompletableFuture < Long > > persist , TimeoutTimer timer ) { Exceptions . checkNotClosed ( this . closed . get ( ) , this ) ; if ( batch . isConditional ( ) ) { // Conditional update. // Collect all Cache Keys for the Update Items that have a condition on them; we need this on order to // serialize execution across them. val keys = batch . getVersionedItems ( ) . stream ( ) . map ( item -> Maps . immutableEntry ( segment . getSegmentId ( ) , item . getHash ( ) ) ) . collect ( Collectors . toList ( ) ) ; // Serialize the execution (queue it up to run only after all other currently queued up conditional updates // for touched keys have finished). return this . conditionalUpdateProcessor . add ( keys , ( ) -> validateConditionalUpdate ( segment , batch , timer ) . thenComposeAsync ( v -> persist . get ( ) , this . executor ) . thenApplyAsync ( batchOffset -> updateCache ( segment , batch , batchOffset ) , this . executor ) ) ; } else { // Unconditional update: persist the entries and update the cache. return persist . get ( ) . thenApplyAsync ( batchOffset -> updateCache ( segment , batch , batchOffset ) , this . executor ) ; } }
Performs a Batch Update or Removal .