fn update_byo_timestamp(&mut self) {
        let mut invalid_byo_sources: Vec<GlobalId> = vec![];
        for (id, byo_consumer) in self.byo_sources.iter_mut() {
            // Get the next set of messages from the Consistency topic
            if let Err(err) = update_source_timestamps(id, &self.tx, byo_consumer) {
                error!(
                    "Failed to correctly parse messages from source {}; {}",
                    id, err
                );
                invalid_byo_sources.push(*id);
            }
        }

        for id in invalid_byo_sources {
            self.drop_source(id);
        }
    }
/// Iterate over each source, updating BYO timestamps
///
/// If an error is encountered while reading the sources, log an error message
/// and remove that source of BYO timestamps
Iterate over each source, updating BYO timestamps
If an error is encountered while reading the sources, log an error message
and remove that source of BYO timestamps
Iterate over each source updating BYO timestamps If an error is encountered while reading the sources log an error message and remove that source of BYO timestamps
fn update_byo_timestamp(&mut self) {
        let mut invalid_byo_sources: Vec<GlobalId> = vec![];
        for (id, byo_consumer) in self.byo_sources.iter_mut() {
            // Get the next set of messages from the Consistency topic
            if let Err(err) = update_source_timestamps(id, &self.tx, byo_consumer) {
                error!(
                    "Failed to correctly parse messages from source {}; {}",
                    id, err
                );
                invalid_byo_sources.push(*id);
            }
        }
        for id in invalid_byo_sources {
            self.drop_source(id);
        }
    }
fn update_byo_timestamp ( & mut self ) { let mut invalid_byo_sources : Vec < GlobalId > = vec ! [ ] ; for ( id , byo_consumer ) in self . byo_sources . iter_mut ( ) { // Get the next set of messages from the Consistency topic if let Err ( err ) = update_source_timestamps ( id , & self . tx , byo_consumer ) { error ! ( " " id err ) ; invalid_byo_sources . push ( * id ) ; } } for id in invalid_byo_sources { self . drop_source ( id ) ; } }
Iterate over each source, updating BYO timestamps
If an error is encountered while reading the sources, log an error message
and remove that source of BYO timestamps
Iterate over each source updating BYO timestamps If an error is encountered while reading the sources log an error message and remove that source of BYO timestamps
&mut self
&mut self