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;
    }
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 ) ; 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 ) ; boolean more = false ; int next = start ; for ( Column < Integer > eventColumn : eventColumns ) { int eventIdx = eventColumn . getName ( ) ; if ( eventIdx == Constants . OPEN_SLAB_MARKER ) { break ; } empty = false ; if ( ! searching ) { more = true ; break ; } EventId eventId = AstyanaxEventId . create ( channel , slabId , eventIdx ) ; ByteBuffer eventData = eventColumn . getByteBufferValue ( ) ; searching = sink . accept ( eventId , eventData ) ; next = eventIdx ; } cursor . set ( next ) ; boolean hasOpenSlabMarker = ! eventColumns . isEmpty ( ) && eventColumns . getColumnByIndex ( eventColumns . size ( ) - 1 ) . getName ( ) == Constants . OPEN_SLAB_MARKER ; boolean stale = open && ! recent && ! hasOpenSlabMarker ; if ( stale ) { _staleSlabMeter . mark ( ) ; } if ( empty && ( ! open || stale ) ) { deleteEmptySlabAsync ( channel , slabId ) ; open = false ; } else if ( stale ) { closeStaleSlabAsync ( channel , slabId ) ; open = false ; } if ( ! more && ! open ) { cursor . set ( SlabCursor . END ) ; } return searching ; }
Returns true to keep searching for more events, false to stop searching for events.