fn update_stats(&mut self) {
        while let Ok(stats) = self.stats_rx.try_recv() {
            if let Some(logger) = self.logger.as_mut() {
                logger.log(MaterializedEvent::KafkaSourceStatistics {
                    source_id: self.id,
                    old: self.last_stats.take(),
                    new: Some(stats.clone()),
                });
                self.last_stats = Some(stats);
            }
        }
    }
/// Read any statistics JSON blobs generated via the rdkafka statistics callback.
Read any statistics JSON blobs generated via the rdkafka statistics callback.
Read any statistics JSON blobs generated via the rdkafka statistics callback .
fn update_stats(&mut self) {
        while let Ok(stats) = self.stats_rx.try_recv() {
            if let Some(logger) = self.logger.as_mut() {
                logger.log(MaterializedEvent::KafkaSourceStatistics {
                    source_id: self.id,
                    old: self.last_stats.take(),
                    new: Some(stats.clone()),
                });
                self.last_stats = Some(stats);
            }
        }
    }
fn update_stats ( & mut self ) { while let Ok ( stats ) = self . stats_rx . try_recv ( ) { if let Some ( logger ) = self . logger . as_mut ( ) { logger . log ( MaterializedEvent :: KafkaSourceStatistics { source_id : self . id , old : self . last_stats . take ( ) , new : Some ( stats . clone ( ) ) , } ) ; self . last_stats = Some ( stats ) ; } } }
Read any statistics JSON blobs generated via the rdkafka statistics callback.
Read any statistics JSON blobs generated via the rdkafka statistics callback .
&mut self
&mut self