private CompletableFuture < Void > processReadResult ( Iterator < Operation > readResult ) { long traceId = LoggerHelpers . traceEnterWithContext ( log , this . traceObjectId , "processReadResult" ) ; InputReadStageResult result = new InputReadStageResult ( this . state ) ; if ( readResult == null ) { // This happens when we get a TimeoutException from the read operation. logStageEvent ( "InputRead" , result ) ; LoggerHelpers . traceLeave ( log , this . traceObjectId , "processReadResult" , traceId ) ; return CompletableFuture . completedFuture ( null ) ; } return Futures . loop ( ( ) -> canRun ( ) && readResult . hasNext ( ) , ( ) -> { Operation op = readResult . next ( ) ; return processOperation ( op ) . thenRun ( ( ) -> { // We have now internalized all operations from this batch; and even if subsequent operations in this iteration // fail, we no longer need to re-read these operations, so update the state with the last read SeqNo. this . state . setLastReadSequenceNumber ( op . getSequenceNumber ( ) ) ; result . operationProcessed ( op ) ; } ) ; } , this . executor ) . thenRun ( ( ) -> { logStageEvent ( "InputRead" , result ) ; LoggerHelpers . traceLeave ( log , this . traceObjectId , "processReadResult" , traceId ) ; } ) ; }
Processes all the operations in the given ReadResult .