void initializeReader ( long initialAllocationDelay ) { boolean alreadyAdded = sync . updateState ( ( state , updates ) -> { if ( state . getSegments ( readerId ) == null ) { log . debug ( "Adding reader {} to reader grop. CurrentState is: {}" , readerId , state ) ; updates . add ( new AddReader ( readerId ) ) ; return false ; } else { return true ; } } ) ; if ( alreadyAdded ) { throw new IllegalStateException ( "The requested reader: " + readerId + " cannot be added to the group because it is already in the group. Perhaps close() was not called?" ) ; } long randomDelay = ( long ) ( Math . random ( ) * Math . min ( initialAllocationDelay , sync . getState ( ) . getConfig ( ) . getGroupRefreshTimeMillis ( ) ) ) ; acquireTimer . reset ( Duration . ofMillis ( initialAllocationDelay + randomDelay ) ) ; }
Add this reader to the reader group so that it is able to acquire segments