fn kafka_get_next_message(
    consumer: &mut BaseConsumer,
) -> Result<Option<BorrowedMessage>, anyhow::Error> {
    if let Some(result) = consumer.poll(Duration::from_millis(60)) {
        match result {
            Ok(message) => Ok(Some(message)),

            Err(err) => {
                bail!("Failed to process message {}", err);
            }
        }
    } else {
        Ok(None)
    }
}
/// Polls a message from a Kafka Source
Polls a message from a Kafka Source
Polls a message from a Kafka Source
fn kafka_get_next_message(
    consumer: &mut BaseConsumer,
) -> Result<Option<BorrowedMessage>, anyhow::Error> {
    if let Some(result) = consumer.poll(Duration::from_millis(60)) {
        match result {
            Ok(message) => Ok(Some(message)),
            Err(err) => {
                bail!("Failed to process message {}", err);
            }
        }
    } else {
        Ok(None)
    }
}
fn kafka_get_next_message ( consumer : & mut BaseConsumer , ) -> Result < Option < BorrowedMessage > , anyhow :: Error > { if let Some ( result ) = consumer . poll ( Duration :: from_millis ( 60 ) ) { match result { Ok ( message ) => Ok ( Some ( message ) ) , Err ( err ) => { bail ! ( " " err ) ; } } } else { Ok ( None ) } }
Polls a message from a Kafka Source
Polls a message from a Kafka Source
consumer
&mut BaseConsumer
consumer
&mut BaseConsumer