private CompletableFuture < Iterator < Operation > > readData ( Void ignored ) { long traceId = LoggerHelpers . traceEnterWithContext ( log , this . traceObjectId , "readData" ) ; try { Duration readTimeout = getReadTimeout ( ) ; return this . dataSource . read ( this . state . getLastReadSequenceNumber ( ) , this . config . getMaxItemsToReadAtOnce ( ) , readTimeout ) . thenApply ( result -> { LoggerHelpers . traceLeave ( log , this . traceObjectId , "readData" , traceId ) ; return result ; } ) . exceptionally ( ex -> { ex = Exceptions . unwrap ( ex ) ; if ( ex instanceof TimeoutException ) { // TimeoutExceptions are acceptable for Reads. In that case we just return null as opposed from // killing the entire Iteration. Even if we were unable to read, we may still need to flush // in this iteration or do other tasks. log . debug ( "{}: Iteration[{}] No items were read during allotted timeout of {}ms" , this . traceObjectId , this . state . getIterationId ( ) , readTimeout . toMillis ( ) ) ; return null ; } else { throw new CompletionException ( ex ) ; } } ) ; } catch ( Throwable ex ) { // This is for synchronous exceptions. Throwable realEx = Exceptions . unwrap ( ex ) ; if ( realEx instanceof TimeoutException ) { logErrorHandled ( realEx ) ; return CompletableFuture . completedFuture ( null ) ; } else { return Futures . failedFuture ( ex ) ; } } }
Reads data from the OperationLog .