boolean handleEndOfSegment ( Segment segmentCompleted ) throws ReaderNotInReaderGroupException { final Map < Segment , List < Long > > segmentToPredecessor ; if ( sync . getState ( ) . getEndSegments ( ) . containsKey ( segmentCompleted ) ) { segmentToPredecessor = Collections . emptyMap ( ) ; } else { val successors = getAndHandleExceptions ( controller . getSuccessors ( segmentCompleted ) , RuntimeException :: new ) ; segmentToPredecessor = successors . getSegmentToPredecessor ( ) ; } AtomicBoolean reinitRequired = new AtomicBoolean ( false ) ; boolean result = sync . updateState ( ( state , updates ) -> { if ( ! state . isReaderOnline ( readerId ) ) { reinitRequired . set ( true ) ; } else { log . debug ( "Marking segment {} as completed in reader group. CurrentState is: {}" , segmentCompleted , state ) ; reinitRequired . set ( false ) ; //This check guards against another checkpoint having started. if ( state . getCheckpointForReader ( readerId ) == null ) { updates . add ( new SegmentCompleted ( readerId , segmentCompleted , segmentToPredecessor ) ) ; return true ; } } return false ; } ) ; if ( reinitRequired . get ( ) ) { throw new ReaderNotInReaderGroupException ( readerId ) ; } acquireTimer . zero ( ) ; return result ; }
Handles a segment being completed by calling the controller to gather all successors to the completed segment . To ensure consistent checkpoints a segment cannot be released while a checkpoint for the reader is pending so it may or may not succeed .