private < T > void addEntry ( StreamElementQueueEntry < T > streamElementQueueEntry ) { assert ( lock . isHeldByCurrentThread ( ) ) ; if ( streamElementQueueEntry . isWatermark ( ) ) { lastSet = new HashSet <> ( capacity ) ; if ( firstSet . isEmpty ( ) ) { firstSet . add ( streamElementQueueEntry ) ; } else { Set < StreamElementQueueEntry < ? > > watermarkSet = new HashSet <> ( 1 ) ; watermarkSet . add ( streamElementQueueEntry ) ; uncompletedQueue . offer ( watermarkSet ) ; } uncompletedQueue . offer ( lastSet ) ; } else { lastSet . add ( streamElementQueueEntry ) ; } streamElementQueueEntry . onComplete ( ( StreamElementQueueEntry < T > value ) -> { try { onCompleteHandler ( value ) ; } catch ( InterruptedException e ) { // The accept executor thread got interrupted. This is probably cause by // the shutdown of the executor. LOG . debug ( "AsyncBufferEntry could not be properly completed because the " + "executor thread has been interrupted." , e ) ; } catch ( Throwable t ) { operatorActions . failOperator ( new Exception ( "Could not complete the " + "stream element queue entry: " + value + ' ' , t ) ) ; } } , executor ) ; numberEntries ++ ; }
Add the given stream element queue entry to the current last set if it is not a watermark . If it is a watermark then stop adding to the current last set insert the watermark into its own set and add a new last set .