fn refresh<S: SourceReader>(
        &mut self,
        source: &mut S,
        timestamp_bindings: &mut TimestampBindingRc,
    ) {
        // Pick up any new partitions that we don't know about but should.
        for (pid, offset) in timestamp_bindings.partitions() {
            if !self.knows_of(&pid) && self.responsible_for(&pid) {
                source.add_partition(pid.clone(), offset.clone());
                self.add_partition(&pid, offset);
            }

            if !self.responsible_for(&pid) {
                continue;
            }
            let cons_info = self
                .partition_metadata
                .get_mut(&pid)
                .expect("partition known to exist");

            if let Some((timestamp, max)) =
                timestamp_bindings.get_binding(&pid, cons_info.offset() + 1)
            {
                cons_info.update_timestamp(timestamp, max);
            }
        }
    }
/// Refreshes the source instance's knowledge of timestamp bindings.
///
/// This function needs to be called at the start of every source operator
/// execution to ensure all source instances assign the same timestamps.
Refreshes the source instance's knowledge of timestamp bindings.
This function needs to be called at the start of every source operator
execution to ensure all source instances assign the same timestamps.
Refreshes the source instance ' s knowledge of timestamp bindings . This function needs to be called at the start of every source operator execution to ensure all source instances assign the same timestamps .
fn refresh<S: SourceReader>(
        &mut self,
        source: &mut S,
        timestamp_bindings: &mut TimestampBindingRc,
    ) {
        // Pick up any new partitions that we don't know about but should.
        for (pid, offset) in timestamp_bindings.partitions() {
            if !self.knows_of(&pid) && self.responsible_for(&pid) {
                source.add_partition(pid.clone(), offset.clone());
                self.add_partition(&pid, offset);
            }
            if !self.responsible_for(&pid) {
                continue;
            }
            let cons_info = self
                .partition_metadata
                .get_mut(&pid)
                .expect("partition known to exist");
            if let Some((timestamp, max)) =
                timestamp_bindings.get_binding(&pid, cons_info.offset() + 1)
            {
                cons_info.update_timestamp(timestamp, max);
            }
        }
    }
fn refresh < S : SourceReader > ( & mut self , source : & mut S , timestamp_bindings : & mut TimestampBindingRc , ) { // Pick up any new partitions that we don't know about but should. for ( pid , offset ) in timestamp_bindings . partitions ( ) { if ! self . knows_of ( & pid ) && self . responsible_for ( & pid ) { source . add_partition ( pid . clone ( ) , offset . clone ( ) ) ; self . add_partition ( & pid , offset ) ; } if ! self . responsible_for ( & pid ) { continue ; } let cons_info = self . partition_metadata . get_mut ( & pid ) . expect ( " " ) ; if let Some ( ( timestamp , max ) ) = timestamp_bindings . get_binding ( & pid , cons_info . offset ( ) + 1 ) { cons_info . update_timestamp ( timestamp , max ) ; } } }
Refreshes the source instance's knowledge of timestamp bindings.
Refreshes the source instance ' s knowledge of timestamp bindings .
&mut self
source
&mut S
timestamp_bindings
&mut TimestampBindingRc
&mut self
source
&mut S
timestamp_bindings
&mut TimestampBindingRc