fn handle_message(
        &mut self,
        message: SourceMessage<Option<Vec<u8>>, Option<Vec<u8>>>,
    ) -> NextMessage<Option<Vec<u8>>, Option<Vec<u8>>> {
        let partition = match message.partition {
            PartitionId::Kafka(pid) => pid,
            _ => unreachable!(),
        };

        // Convert the received offset back from a 1-indexed MzOffset to the correct offset.
        let offset = message.offset.offset - 1;
        // Offsets are guaranteed to be 1) monotonically increasing *unless* there is
        // a network issue or a new partition added, at which point the consumer may
        // start processing the topic from the beginning, or we may see duplicate offsets
        // At all times, the guarantee : if we see offset x, we have seen all offsets [0,x-1]
        // that we are ever going to see holds.
        // Offsets are guaranteed to be contiguous when compaction is disabled. If compaction
        // is enabled, there may be gaps in the sequence.
        // If we see an "old" offset, we ast-forward the consumer and skip that message

        // Given the explicit consumer to partition assignment, we should never receive a message
        // for a partition for which we have no metadata
        assert!(self.last_offsets.contains_key(&partition));

        let last_offset_ref = self
            .last_offsets
            .get_mut(&partition)
            .expect("partition known to be installed");

        let last_offset = *last_offset_ref;
        if offset <= last_offset {
            info!(
                "Kafka message before expected offset, skipping: \
                             source {} (reading topic {}, partition {}) \
                             received offset {} expected offset {:?}",
                self.source_name,
                self.topic_name,
                partition,
                offset,
                last_offset + 1,
            );
            // Seek to the *next* offset (aka last_offset + 1) that we have not yet processed
            self.fast_forward_consumer(partition, last_offset + 1);
            // We explicitly should not consume the message as we have already processed it
            // However, we make sure to activate the source to make sure that we get a chance
            // to read from this consumer again (even if no new data arrives)
            NextMessage::TransientDelay
        } else {
            *last_offset_ref = offset;
            NextMessage::Ready(message)
        }
    }
/// Checks if the given message is viable for emission. This checks if the message offset is
/// past the expected offset and seeks the consumer if it is not.
Checks if the given message is viable for emission. This checks if the message offset is
past the expected offset and seeks the consumer if it is not.
Checks if the given message is viable for emission . This checks if the message offset is past the expected offset and seeks the consumer if it is not .
fn handle_message(
        &mut self,
        message: SourceMessage<Option<Vec<u8>>, Option<Vec<u8>>>,
    ) -> NextMessage<Option<Vec<u8>>, Option<Vec<u8>>> {
        let partition = match message.partition {
            PartitionId::Kafka(pid) => pid,
            _ => unreachable!(),
        };
        // Convert the received offset back from a 1-indexed MzOffset to the correct offset.
        let offset = message.offset.offset - 1;
        // Offsets are guaranteed to be 1) monotonically increasing *unless* there is
        // a network issue or a new partition added, at which point the consumer may
        // start processing the topic from the beginning, or we may see duplicate offsets
        // At all times, the guarantee : if we see offset x, we have seen all offsets [0,x-1]
        // that we are ever going to see holds.
        // Offsets are guaranteed to be contiguous when compaction is disabled. If compaction
        // is enabled, there may be gaps in the sequence.
        // If we see an "old" offset, we ast-forward the consumer and skip that message
        // Given the explicit consumer to partition assignment, we should never receive a message
        // for a partition for which we have no metadata
        assert!(self.last_offsets.contains_key(&partition));
        let last_offset_ref = self
            .last_offsets
            .get_mut(&partition)
            .expect("partition known to be installed");
        let last_offset = *last_offset_ref;
        if offset <= last_offset {
            info!(
                "Kafka message before expected offset, skipping: \
                             source {} (reading topic {}, partition {}) \
                             received offset {} expected offset {:?}",
                self.source_name,
                self.topic_name,
                partition,
                offset,
                last_offset + 1,
            );
            // Seek to the *next* offset (aka last_offset + 1) that we have not yet processed
            self.fast_forward_consumer(partition, last_offset + 1);
            // We explicitly should not consume the message as we have already processed it
            // However, we make sure to activate the source to make sure that we get a chance
            // to read from this consumer again (even if no new data arrives)
            NextMessage::TransientDelay
        } else {
            *last_offset_ref = offset;
            NextMessage::Ready(message)
        }
    }
fn handle_message ( & mut self , message : SourceMessage < Option < Vec < u8 > > , Option < Vec < u8 > > > , ) -> NextMessage < Option < Vec < u8 > > , Option < Vec < u8 > > > { let partition = match message . partition { PartitionId :: Kafka ( pid ) => pid , _ => unreachable ! ( ) , } ; // Convert the received offset back from a 1-indexed MzOffset to the correct offset. let offset = message . offset . offset - 1 ; // Offsets are guaranteed to be 1) monotonically increasing *unless* there is // a network issue or a new partition added, at which point the consumer may // start processing the topic from the beginning, or we may see duplicate offsets // At all times, the guarantee : if we see offset x, we have seen all offsets [0,x-1] // that we are ever going to see holds. // Offsets are guaranteed to be contiguous when compaction is disabled. If compaction // is enabled, there may be gaps in the sequence. // If we see an "old" offset, we ast-forward the consumer and skip that message // Given the explicit consumer to partition assignment, we should never receive a message // for a partition for which we have no metadata assert ! ( self last_offsets contains_key ( partition ) ) ; let last_offset_ref = self . last_offsets . get_mut ( & partition ) . expect ( " " ) ; let last_offset = * last_offset_ref ; if offset <= last_offset { info ! ( " \
 \
 " self source_name self topic_name partition offset last_offset 1 ) ; // Seek to the *next* offset (aka last_offset + 1) that we have not yet processed self . fast_forward_consumer ( partition , last_offset + 1 ) ; // We explicitly should not consume the message as we have already processed it // However, we make sure to activate the source to make sure that we get a chance // to read from this consumer again (even if no new data arrives) NextMessage :: TransientDelay } else { * last_offset_ref = offset ; NextMessage :: Ready ( message ) } }
Checks if the given message is viable for emission.
Checks if the given message is viable for emission .
&mut self
message
SourceMessage<Option<Vec<u8>>, Option<Vec<u8>>>
&mut self
message
SourceMessage<Option<Vec<u8>>, Option<Vec<u8>>>