Struct FilteredReader 

pub struct FilteredReader { /* private fields */ }

Reads the records a consumer group’s policy selects, with their original offsets, and stores progress through fenced acknowledgments.

Pages are read ahead of acknowledgments. An acknowledgment stores a page’s safe offset only after that page and every earlier page of the partition are fully handled, so progress never skips an unhandled record. Pages that matched nothing are stored by the reader itself.

A partition that fails, faults, or blocks waits one idle interval before it is read again, while the other partitions keep reading. A purge or a new group binding restarts the partition from its stored offset once, and the error that reports it reaches the caller.

Every method is cancel-safe: a cancelled call leaves the position and the unstored progress as they were, and the next call repeats the work.

Implementations

impl FilteredReader

pub async fn next_page(&mut self) -> Result<MatchedPage, LaserError>

The next page with at least one matching record. Waits while every partition has nothing new. Wrap it in a timeout to bound the wait.

pub async fn try_next_page(&mut self) -> Result<Option<MatchedPage>, LaserError>

Read each partition in turn until a page matches or every partition has nothing new, without the idle wait. None when nothing is new, so a caller that drives its own loop waits idle_interval before it asks again.

pub async fn read_round( &mut self, ) -> Result<(Option<MatchedPage>, bool), LaserError>

Perform one bounded round without an idle wait. The flag means unmatched work remains. Read examined_in_round for its source-record count, including pages that matched nothing.

pub const fn examined_in_round(&self) -> u64

Source records examined by the last bounded round, including empty pages that advance progress.

pub const fn idle_interval(&self) -> Duration

How long this reader waits when nothing is new.

pub fn owns(&self, page: &MatchedPage) -> bool

Whether page came from this reader in its current membership, so its records can still be acknowledged. A rejoin retires every earlier page.

pub async fn next_record(&mut self) -> Result<MatchedRecord, LaserError>

The next matching record, one at a time over next_page.

pub async fn ack(&mut self, record: &MatchedRecord) -> Result<(), LaserError>

Mark one record handled, and store the progress this completes.

pub async fn ack_through( &mut self, record: &MatchedRecord, ) -> Result<(), LaserError>

Mark every record on this partition through record handled, and store that completed prefix. Process all preceding records first. Records after it remain pending, even inside the same page.

pub async fn ack_page(&mut self, page: &MatchedPage) -> Result<(), LaserError>

Mark every record of page handled, and store the progress this completes.

pub fn data_connections_opened(&self) -> u64

Data connections this reader opened to partition primaries. A healthy reader opens one per node and keeps it: the count grows only when a node stops being primary, a connection fails, or the group session ends. A local reader opens none.

pub fn partitions(&self) -> Vec<u32>

The partitions this reader reads now.

pub async fn close(self) -> Result<(), LaserError>

Store completed progress, leave the group, and close the data connections. Handled records that were not acknowledged are read again by the next reader. A reader dropped without close leaves its group when its own coordinator connection closes.