private void checkValidStorageOperation ( StorageOperation operation ) throws DataCorruptionException { // StreamSegmentAppendOperations need to be pre-processed into CachedStreamSegmentAppendOperations. Preconditions . checkArgument ( ! ( operation instanceof StreamSegmentAppendOperation ) , "SegmentAggregator cannot process StreamSegmentAppendOperations." ) ; // Verify operation offset against the lastAddedOffset (whether the last Op in the list or StorageLength). long offset = operation . getStreamSegmentOffset ( ) ; long length = operation . getLength ( ) ; Preconditions . checkArgument ( offset >= 0 , "Operation '%s' has an invalid offset (%s)." , operation , operation . getStreamSegmentOffset ( ) ) ; Preconditions . checkArgument ( length >= 0 , "Operation '%s' has an invalid length (%s)." , operation , operation . getLength ( ) ) ; // Check that operations are contiguous (only for the operations after the first one - as we initialize lastAddedOffset on the first op). if ( isTruncateOperation ( operation ) ) { if ( this . metadata . getStartOffset ( ) < operation . getStreamSegmentOffset ( ) ) { throw new DataCorruptionException ( String . format ( "StreamSegmentTruncateOperation '%s' has a truncation offset beyond the one in the Segment's Metadata. Expected: at most %d, actual: %d." , operation , this . metadata . getStartOffset ( ) , offset ) ) ; } } else { long lastOffset = this . lastAddedOffset . get ( ) ; if ( lastOffset >= 0 && offset != lastOffset ) { throw new DataCorruptionException ( String . format ( "Wrong offset for Operation '%s'. Expected: %s, actual: %d." , operation , this . lastAddedOffset , offset ) ) ; } } // Check that the operation does not exceed the Length of the StreamSegment. if ( offset + length > this . metadata . getLength ( ) ) { throw new DataCorruptionException ( String . format ( "Operation '%s' has at least one byte beyond its Length. Offset = %d, Length = %d, Length = %d." , operation , offset , length , this . metadata . getLength ( ) ) ) ; } if ( operation instanceof StreamSegmentSealOperation ) { // For StreamSegmentSealOperations, we must ensure the offset of the operation is equal to the Length for the segment. if ( this . metadata . getLength ( ) != offset ) { throw new DataCorruptionException ( String . format ( "Wrong offset for Operation '%s'. Expected: %d (Length), actual: %d." , operation , this . metadata . getLength ( ) , offset ) ) ; } // Even though not an offset, we should still verify that the metadata actually thinks this is a sealed segment. if ( ! this . metadata . isSealed ( ) ) { throw new DataCorruptionException ( String . format ( "Received Operation '%s' for a non-sealed segment." , operation ) ) ; } } }
Validates that the given StorageOperation can be processed given the current accumulated state of the Segment .