private CompletableFuture < Integer > reconcileData ( AggregatedAppendOperation op , SegmentProperties storageInfo , TimeoutTimer timer ) { InputStream appendStream = this . dataSource . getAppendData ( op . getStreamSegmentId ( ) , op . getStreamSegmentOffset ( ) , ( int ) op . getLength ( ) ) ; if ( appendStream == null ) { return Futures . failedFuture ( new ReconciliationFailureException ( String . format ( "Unable to reconcile operation '%s' because no append data is associated with it." , op ) , this . metadata , storageInfo ) ) ; } // Only read as much data as we need. long readLength = Math . min ( op . getLastStreamSegmentOffset ( ) , storageInfo . getLength ( ) ) - op . getStreamSegmentOffset ( ) ; assert readLength > 0 : "Append Operation to be reconciled is beyond the Segment's StorageLength (" + storageInfo . getLength ( ) + "): " + op ; // Read all data from storage. byte [ ] storageData = new byte [ ( int ) readLength ] ; AtomicInteger reconciledBytes = new AtomicInteger ( ) ; return Futures . loop ( ( ) -> reconciledBytes . get ( ) < readLength , ( ) -> this . storage . read ( this . handle . get ( ) , op . getStreamSegmentOffset ( ) + reconciledBytes . get ( ) , storageData , reconciledBytes . get ( ) , ( int ) readLength - reconciledBytes . get ( ) , timer . getRemaining ( ) ) , bytesRead -> { assert bytesRead > 0 : String . format ( "Unable to make any read progress when reconciling operation '%s' after reading %s bytes." , op , reconciledBytes ) ; reconciledBytes . addAndGet ( bytesRead ) ; } , this . executor ) . thenApplyAsync ( v -> { // Compare, byte-by-byte, the contents of the append. verifySame ( appendStream , storageData , op , storageInfo ) ; return reconciledBytes . get ( ) ; } , this . executor ) ; }
Attempts to reconcile the data for the given AggregatedAppendOperation . Since Append Operations can be partially flushed reconciliation may be for the full operation or for a part of it .