fn fast_forward_consumer(&self, pid: i32, next_offset: i64) {
        let res = self.consumer.seek(
            &self.topic_name,
            pid,
            Offset::Offset(next_offset),
            Duration::from_secs(1),
        );
        match res {
            Ok(_) => {
                let res = self.consumer.position().unwrap_or_default().to_topic_map();
                let position = res
                    .get(&(self.topic_name.clone(), pid))
                    .and_then(|p| match p {
                        Offset::Offset(o) => Some(o),
                        _ => None,
                    });
                if let Some(position) = position {
                    if *position != next_offset {
                        warn!("Did not fast-forward consumer on partition PID: {} to the correct Kafka offset. Currently at offset: {} Expected offset: {}",
                              pid, position, next_offset);
                    } else {
                        info!("Successfully fast-forwarded consumer on partition PID: {} to Kafka offset {}.", pid, position);
                    }
                } else {
                    warn!("Tried to fast-forward consumer on partition PID: {} to Kafka offset {}. Could not obtain new consumer position",
                          pid, next_offset);
                }
            }
            Err(e) => error!(
                "Failed to fast-forward consumer for source:{}, Error:{}",
                self.source_name, e
            ),
        }
    }
/// Fast-forward consumer to specified Kafka Offset. Prints a warning if failed to do so
/// Assumption: if offset does not exist (for instance, because of compaction), will seek
/// to the next available offset
Fast-forward consumer to specified Kafka Offset. Prints a warning if failed to do so
Assumption: if offset does not exist (for instance, because of compaction), will seek
to the next available offset
Fast - forward consumer to specified Kafka Offset . Prints a warning if failed to do so Assumption : if offset does not exist ( for instance because of compaction ) will seek to the next available offset
fn fast_forward_consumer(&self, pid: i32, next_offset: i64) {
        let res = self.consumer.seek(
            &self.topic_name,
            pid,
            Offset::Offset(next_offset),
            Duration::from_secs(1),
        );
        match res {
            Ok(_) => {
                let res = self.consumer.position().unwrap_or_default().to_topic_map();
                let position = res
                    .get(&(self.topic_name.clone(), pid))
                    .and_then(|p| match p {
                        Offset::Offset(o) => Some(o),
                        _ => None,
                    });
                if let Some(position) = position {
                    if *position != next_offset {
                        warn!("Did not fast-forward consumer on partition PID: {} to the correct Kafka offset. Currently at offset: {} Expected offset: {}",
                              pid, position, next_offset);
                    } else {
                        info!("Successfully fast-forwarded consumer on partition PID: {} to Kafka offset {}.", pid, position);
                    }
                } else {
                    warn!("Tried to fast-forward consumer on partition PID: {} to Kafka offset {}. Could not obtain new consumer position",
                          pid, next_offset);
                }
            }
            Err(e) => error!(
                "Failed to fast-forward consumer for source:{}, Error:{}",
                self.source_name, e
            ),
        }
    }
fn fast_forward_consumer ( & self , pid : i32 , next_offset : i64 ) { let res = self . consumer . seek ( & self . topic_name , pid , Offset :: Offset ( next_offset ) , Duration :: from_secs ( 1 ) , ) ; match res { Ok ( _ ) => { let res = self . consumer . position ( ) . unwrap_or_default ( ) . to_topic_map ( ) ; let position = res . get ( & ( self . topic_name . clone ( ) , pid ) ) . and_then ( | p | match p { Offset :: Offset ( o ) => Some ( o ) , _ => None , } ) ; if let Some ( position ) = position { if * position != next_offset { warn ! ( " " pid position next_offset ) ; } else { info ! ( " " pid position ) ; } } else { warn ! ( " " pid next_offset ) ; } } Err ( e ) => error ! ( " " self source_name e ) , } }
Fast-forward consumer to specified Kafka Offset.
Fast - forward consumer to specified Kafka Offset .
&self
pid
i32
next_offset
i64
&self
pid
i32
next_offset
i64