[+] Add async processing in kafka

Author: aofdev

kafka/Cargo.toml

@@ -10,14 +10,22 @@ edition = "2018"
 tokio = { version = "1.8.1", features = ["full"] }
 clap = "2.33.3"
 log = "0.4.14"
+rand = "0.8.4"
+futures = "0.3.15"
 chrono = "0.4.19"
 env_logger = "0.9.0"
 rdkafka = { version = "0.26", features = ["cmake-build"] }
+reqwest = { version = "0.11", features = ["blocking", "json"] }
+serde = { version = "1.0.126", features = ["derive"] }
 
 [[bin]]
 name = "producer"
 path = "src/producer.rs"
 
 [[bin]]
 name = "consumer"
-path = "src/consumer.rs"
+path = "src/consumer.rs"
+
+[[bin]]
+name = "async_processing"
+path = "src/async_processing.rs"

kafka/src/async_processing.rs

@@ -0,0 +1,192 @@
+use clap::{value_t, App, Arg};
+use futures::stream::FuturesUnordered;
+use futures::{StreamExt, TryStreamExt};
+use log::info;
+use rand::Rng;
+use rdkafka::config::ClientConfig;
+use rdkafka::consumer::stream_consumer::StreamConsumer;
+use rdkafka::consumer::Consumer;
+use rdkafka::message::{BorrowedMessage, OwnedMessage};
+use rdkafka::producer::{FutureProducer, FutureRecord};
+use rdkafka::Message;
+use serde::Deserialize;
+use std::time::Duration;
+
+use crate::utils::setup_logger;
+
+mod utils;
+
+#[derive(Deserialize, Debug)]
+struct ResponseJSONPlaceholder {
+    title: String,
+}
+
+fn fetch_json_placeholder() -> Result<String, Box<dyn std::error::Error>> {
+    let num = rand::thread_rng().gen_range(0..100);
+    let url = format!("https://jsonplaceholder.typicode.com/todos/{}", &num);
+    let resp = reqwest::blocking::Client::new()
+        .get(url)
+        .send()?
+        .json::<ResponseJSONPlaceholder>()?;
+
+    Ok(resp.title)
+}
+
+async fn record_borrowed_message_receipt(msg: &BorrowedMessage<'_>) {
+    // Simulate some work that must be done in the same order as messages are
+    // received; i.e., before truly parallel processing can begin.
+    info!("Message received: {}", msg.offset());
+}
+
+async fn record_owned_message_receipt(_msg: &OwnedMessage) {
+    // Like `record_borrowed_message_receipt`, but takes an `OwnedMessage`
+    // instead, as in a real-world use case  an `OwnedMessage` might be more
+    // convenient than a `BorrowedMessage`.
+}
+
+fn get_title(msg: OwnedMessage) -> String {
+    match msg.payload_view::<str>() {
+        Some(Ok(_)) => match fetch_json_placeholder() {
+            Ok(title) => format!("Payload title: {}", title),
+            Err(_) => "Error: fetch_json_placeholder somethings".to_owned(),
+        },
+        Some(Err(_)) => "Message payload is not a string".to_owned(),
+        None => "No payload".to_owned(),
+    }
+}
+
+async fn run_async_processor(
+    brokers: String,
+    group_id: String,
+    input_topic: String,
+    output_topic: String,
+) {
+    let consumer: StreamConsumer = ClientConfig::new()
+        .set("group.id", &group_id)
+        .set("bootstrap.servers", &brokers)
+        .set("enable.partition.eof", "false")
+        .set("session.timeout.ms", "6000")
+        .set("enable.auto.commit", "false")
+        .create()
+        .expect("Consumer creation failed");
+
+    consumer
+        .subscribe(&[&input_topic])
+        .expect("Can't subscribe to specified topic");
+
+    // Create the `FutureProducer` to produce asynchronously.
+    let producer: FutureProducer = ClientConfig::new()
+        .set("bootstrap.servers", &brokers)
+        .set("message.timeout.ms", "5000")
+        .create()
+        .expect("Producer creation error");
+
+    // Create the outer pipeline on the message stream.
+    let stream_processor = consumer.stream().try_for_each(|borrowed_message| {
+        let producer = producer.clone();
+        let output_topic = output_topic.to_string();
+        async move {
+            // Process each message
+            record_borrowed_message_receipt(&borrowed_message).await;
+            // Borrowed messages can't outlive the consumer they are received from, so they need to
+            // be owned in order to be sent to a separate thread.
+            let owned_message = borrowed_message.detach();
+            record_owned_message_receipt(&owned_message).await;
+            tokio::spawn(async move {
+                // The body of this block will be executed on the main thread pool,
+                // but we perform `get_title` on a separate thread pool
+                // for CPU-intensive tasks via `tokio::task::spawn_blocking`.
+                let computation_result = tokio::task::spawn_blocking(|| get_title(owned_message))
+                    .await
+                    .expect("failed to wait for expensive computation");
+                let produce_future = producer.send(
+                    FutureRecord::to(&output_topic)
+                        .key("some key")
+                        .payload(&computation_result),
+                    Duration::from_secs(0),
+                );
+                match produce_future.await {
+                    Ok(delivery) => println!("Sent: {:?}", delivery),
+                    Err((e, _)) => println!("Error: {:?}", e),
+                }
+            });
+            Ok(())
+        }
+    });
+
+    info!("Starting event loop");
+    stream_processor.await.expect("stream processing failed");
+    info!("Stream processing terminated");
+}
+
+#[tokio::main]
+async fn main() {
+    let matches = App::new("Async example")
+        .version(option_env!("CARGO_PKG_VERSION").unwrap_or(""))
+        .about("Asynchronous processing example")
+        .arg(
+            Arg::with_name("brokers")
+                .short("b")
+                .long("brokers")
+                .help("Broker list in kafka format")
+                .takes_value(true)
+                .default_value("localhost:9092"),
+        )
+        .arg(
+            Arg::with_name("group-id")
+                .short("g")
+                .long("group-id")
+                .help("Consumer group id")
+                .takes_value(true)
+                .default_value("consumer_group_id"),
+        )
+        .arg(
+            Arg::with_name("log-conf")
+                .long("log-conf")
+                .help("Configure the logging format (example: 'rdkafka=trace')")
+                .takes_value(true),
+        )
+        .arg(
+            Arg::with_name("input-topic")
+                .long("input-topic")
+                .help("Input topic")
+                .takes_value(true)
+                .required(true),
+        )
+        .arg(
+            Arg::with_name("output-topic")
+                .long("output-topic")
+                .help("Output topic")
+                .takes_value(true)
+                .required(true),
+        )
+        .arg(
+            Arg::with_name("num-workers")
+                .long("num-workers")
+                .help("Number of workers")
+                .takes_value(true)
+                .default_value("1"),
+        )
+        .get_matches();
+
+    setup_logger(true, matches.value_of("log-conf"));
+
+    let brokers = matches.value_of("brokers").unwrap();
+    let group_id = matches.value_of("group-id").unwrap();
+    let input_topic = matches.value_of("input-topic").unwrap();
+    let output_topic = matches.value_of("output-topic").unwrap();
+    let num_workers = value_t!(matches, "num-workers", usize).unwrap();
+
+    (0..num_workers)
+        .map(|_| {
+            tokio::spawn(run_async_processor(
+                brokers.to_owned(),
+                group_id.to_owned(),
+                input_topic.to_owned(),
+                output_topic.to_owned(),
+            ))
+        })
+        .collect::<FuturesUnordered<_>>()
+        .for_each(|_| async { () })
+        .await
+}