fn add_partition(&mut self, pid: &PartitionId, offset: Option<MzOffset>) {
        if self.partition_metadata.contains_key(pid) {
            error!("Incorrectly attempting to add a partition twice for source: {} partition: {}. Ignoring",
                   self.source_id,
                   pid
            );

            return;
        }

        self.partition_metadata
            .insert(pid.clone(), ConsInfo::new(self.last_closed_ts, offset));
        self.source_metrics.add_partition(pid);
    }
/// Start tracking consistency information and metrics for `pid`. The optional `offset` will be
/// used to correctly set the internal read offset, though sources are responsible for correctly
/// starting to read only from that offset.
///
/// Need to call this together with `SourceReader::add_partition` before we
/// ingest from `pid`.
Start tracking consistency information and metrics for `pid`. The optional `offset` will be
used to correctly set the internal read offset, though sources are responsible for correctly
starting to read only from that offset.

Need to call this together with `SourceReader::add_partition` before we
ingest from `pid`.
Start tracking consistency information and metrics for ` pid ` . The optional ` offset ` will be used to correctly set the internal read offset though sources are responsible for correctly starting to read only from that offset . Need to call this together with ` SourceReader :: add_partition ` before we ingest from ` pid ` .
fn add_partition(&mut self, pid: &PartitionId, offset: Option<MzOffset>) {
        if self.partition_metadata.contains_key(pid) {
            error!("Incorrectly attempting to add a partition twice for source: {} partition: {}. Ignoring",
                   self.source_id,
                   pid
            );
            return;
        }
        self.partition_metadata
            .insert(pid.clone(), ConsInfo::new(self.last_closed_ts, offset));
        self.source_metrics.add_partition(pid);
    }
fn add_partition ( & mut self , pid : & PartitionId , offset : Option < MzOffset > ) { if self . partition_metadata . contains_key ( pid ) { error ! ( " " self source_id pid ) ; return ; } self . partition_metadata . insert ( pid . clone ( ) , ConsInfo :: new ( self . last_closed_ts , offset ) ) ; self . source_metrics . add_partition ( pid ) ; }
Start tracking consistency information and metrics for `pid`.
Start tracking consistency information and metrics for ` pid ` .
&mut self
pid
&PartitionId
offset
Option<MzOffset>
&mut self
pid
&PartitionId
offset
Option<MzOffset>