The consumer_app
is a simple Kafka consumer written in Rust. It listens to messages from a Kafka topic.
- Rust
- Docker & Docker Compose (optional to run Kafka Services)
-
Clone the Repository
git clone git@github.com:gustavobrossi/consumer_app.git cd consumer_app
-
Build the Application
cargo build --release
-
Run the Kafka Services
Make sure you have a Kafka Services running so the application can connect and consume from it. You will only need one server that both producer and consumer will connect into. For testing puropse, use the provided
docker-compose.yml
file bellow to set up Kafka.version: '3' services: zookeeper: image: zookeeper ports: - "2181:2181" kafka: image: bitnami/kafka ports: - "9092:9092" environment: KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092 KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
docker-compose up -d
-
Run the Consumer Application
cargo run --release
-
main.rs
This file sets up a Kafka consumer and listens for messages from a topic.
use rdkafka::consumer::{Consumer, StreamConsumer}; use rdkafka::config::ClientConfig; use rdkafka::message::BorrowedMessage; use rdkafka::util::get_rdkafka_version; use futures::StreamExt; use rdkafka::Message; async fn consume_message(message: &BorrowedMessage<'_>) { let payload = match message.payload_view::<str>() { Some(Ok(payload)) => payload, Some(Err(_)) => "<invalid utf-8>", None => "<null>", }; println!( "key: '{:?}', payload: '{}', topic: {}, partition: {}, offset: {}", message.key(), payload, message.topic(), message.partition(), message.offset() ); } #[tokio::main] async fn main() { let (version_n, version_s) = get_rdkafka_version(); println!("rd_kafka_version: 0x{:08x}, {}", version_n, version_s); let consumer: StreamConsumer = ClientConfig::new() .set("group.id", "example_consumer_group") .set("bootstrap.servers", "localhost:9092") .set("enable.partition.eof", "false") .set("auto.offset.reset", "earliest") .create() .expect("Consumer creation failed"); consumer.subscribe(&["test_topic"]) .expect("Can't subscribe to specified topics"); let mut message_stream = consumer.stream(); while let Some(message) = message_stream.next().await { match message { Ok(m) => consume_message(&m).await, Err(e) => eprintln!("Kafka error: {}", e), } } }
-
cargo.toml
[package] name = "consumer_app" version = "0.1.0" edition = "2021" [dependencies] rdkafka = "0.26" tokio = { version = "1", features = ["full"] } futures = "0.3"