private boolean readSlab ( String channel , ByteBuffer slabId , SlabCursor cursor , boolean open , EventSink sink ) { int start = cursor . get ( ) ; if ( start == SlabCursor . END ) { return true ; } boolean recent = isRecent ( slabId ) ; // Event add and delete write with local quorum, so read with local quorum to get a consistent view of things. // Using a lower consistency level could result in (a) duplicate events because we miss deletes and (b) // incorrectly closing or deleting slabs when slabs look empty if we miss adds. ColumnList < Integer > eventColumns = execute ( _keyspace . prepareQuery ( ColumnFamilies . SLAB , ConsistencyLevel . CL_LOCAL_QUORUM ) . getKey ( slabId ) . withColumnRange ( start , Constants . OPEN_SLAB_MARKER , false , Integer . MAX_VALUE ) ) ; boolean searching = true ; boolean empty = ( start == 0 ) ; // If we skipped events in the query we must assume the slab isn't empty. boolean more = false ; int next = start ; for ( Column < Integer > eventColumn : eventColumns ) { int eventIdx = eventColumn . getName ( ) ; // Open slabs have a dummy entry at maxint that indicates that this slab is still open. if ( eventIdx == Constants . OPEN_SLAB_MARKER ) { break ; } // Found at least one data item. empty = false ; if ( ! searching ) { more = true ; // There are more events to be found next time we poll this slab. break ; } // Pass the data on to the EventSink. It will tell us whether or not to keep searching. EventId eventId = AstyanaxEventId . create ( channel , slabId , eventIdx ) ; ByteBuffer eventData = eventColumn . getByteBufferValue ( ) ; searching = sink . accept ( eventId , eventData ) ; next = eventIdx ; } // Next time we query this slab start the search with last event received by the sink, repeating it. cursor . set ( next ) ; // Stale open slab? Rare, should only happen when a writer crashes without cleaning up and closing its open // slabs. Normally writers re-write the OPEN_SLAB_MARKER column on every write as a sort of heartbeat. Readers // detect "stale" slabs when the open slab markers expire, and they close those slabs on behalf of the crashed writers. boolean hasOpenSlabMarker = ! eventColumns . isEmpty ( ) && eventColumns . getColumnByIndex ( eventColumns . size ( ) - 1 ) . getName ( ) == Constants . OPEN_SLAB_MARKER ; boolean stale = open && ! recent && ! hasOpenSlabMarker ; if ( stale ) { _staleSlabMeter . mark ( ) ; } // If the slab is currently closed or should be closed then it will never receive more data so check to see if // we can (a) delete it (it's empty) or at least (b) close it. if ( empty && ( ! open || stale ) ) { deleteEmptySlabAsync ( channel , slabId ) ; open = false ; } else if ( stale ) { closeStaleSlabAsync ( channel , slabId ) ; open = false ; } // If we ran through all the data in a closed slab, skip this slab next time. This is especially common with // badly-behaving Databus listeners that poll repeatedly but don't ack. if ( ! more && ! open ) { cursor . set ( SlabCursor . END ) ; } return searching ; }
Returns true to keep searching for more events false to stop searching for events .