fn poll_from_next_queue(&mut self) -> Option<SourceMessage<Option<Vec<u8>>, Option<Vec<u8>>>> {
        let mut partition_queue = self.partition_consumers.pop_front().unwrap();

        let message = match partition_queue.get_next_message() {
            Err(e) => {
                let pid = partition_queue.pid();
                let last_offset = self
                    .last_offsets
                    .get(&pid)
                    .expect("partition known to be installed");

                error!(
                        "kafka error consuming from source: {} topic: {}: partition: {} last processed offset: {} : {}",
                        self.source_name,
                        self.topic_name,
                        pid,
                        last_offset,
                        e
                    );
                None
            }
            Ok(m) => m,
        };

        self.partition_consumers.push_back(partition_queue);

        message
    }
/// Polls from the next partition queue and returns the message, if any.
///
/// We maintain the list of partition queues in a queue, and add queues that we polled from to
/// the end of the queue. We thus swing through all available partition queues in a somewhat
/// fair manner.
Polls from the next partition queue and returns the message, if any.
We maintain the list of partition queues in a queue, and add queues that we polled from to
the end of the queue. We thus swing through all available partition queues in a somewhat
fair manner.
Polls from the next partition queue and returns the message if any . We maintain the list of partition queues in a queue and add queues that we polled from to the end of the queue . We thus swing through all available partition queues in a somewhat fair manner .
fn poll_from_next_queue(&mut self) -> Option<SourceMessage<Option<Vec<u8>>, Option<Vec<u8>>>> {
        let mut partition_queue = self.partition_consumers.pop_front().unwrap();
        let message = match partition_queue.get_next_message() {
            Err(e) => {
                let pid = partition_queue.pid();
                let last_offset = self
                    .last_offsets
                    .get(&pid)
                    .expect("partition known to be installed");
                error!(
                        "kafka error consuming from source: {} topic: {}: partition: {} last processed offset: {} : {}",
                        self.source_name,
                        self.topic_name,
                        pid,
                        last_offset,
                        e
                    );
                None
            }
            Ok(m) => m,
        };
        self.partition_consumers.push_back(partition_queue);
        message
    }
fn poll_from_next_queue ( & mut self ) -> Option < SourceMessage < Option < Vec < u8 > > , Option < Vec < u8 > > > > { let mut partition_queue = self . partition_consumers . pop_front ( ) . unwrap ( ) ; let message = match partition_queue . get_next_message ( ) { Err ( e ) => { let pid = partition_queue . pid ( ) ; let last_offset = self . last_offsets . get ( & pid ) . expect ( " " ) ; error ! ( " " self source_name self topic_name pid last_offset e ) ; None } Ok ( m ) => m , } ; self . partition_consumers . push_back ( partition_queue ) ; message }
Polls from the next partition queue and returns the message, if any.
Polls from the next partition queue and returns the message if any .
&mut self
&mut self