Code review to handle message commit within an asynchronous task and semaphores

[dertin]

Hi,

I'm seeking assistance with my Rust code and would appreciate a code review. The current implementation involves handling message commits within asynchronous tasks, with the additional complexity of using semaphores to control the consumption of new messages and manage the number of concurrently running asynchronous tasks. I'm particularly interested in any suggestions or feedback on improving the efficiency and robustness of this design.

Thank you in advance for your insights!

rdkafka::ClientConfig::new()
.set("bootstrap.servers", kafka_brokers)
.set("group.id", kafka_group)
.set("enable.partition.eof", "false")
.set("session.timeout.ms", "6000")
.set("enable.auto.commit", "false")
.set("auto.offset.reset", "earliest")
.set("sasl.username", username)
.set("sasl.password", password)
.set("sasl.mechanisms", "SCRAM-SHA-256")
.set("security.protocol", "SASL_SSL");

---

let semaphore_arc = Arc::new(Semaphore::new(15000));
let consumer_arc = Arc::new(consumer);

loop {
    
    let semaphore_clone = Arc::clone(&semaphore_arc);
    
    let semaphore_permit = match semaphore_clone.acquire_owned().await {
        Ok(permit) => permit,
        Err(e) => {
            println!("Error acquiring semaphore: {}", e);
            continue;
        }
    };

    let consumer_clone = Arc::clone(&consumer_arc);
    
    match consumer_arc.recv().await {
        Err(e) => println!("Error from consumer: {}", e),
        Ok(borrowed_message) => {
            
            // Deserialize the message payload.
            
            let payload = match borrowed_message.payload_view::<str>() {
                None => "",
                Some(Ok(s)) => s,
                Some(Err(e)) => {
                    
                    // If the message is not valid UTF-8, commit the message offset and release the semaphore permit.
                    // I don't want to process this message again.

                    println!("Error while deserializing message payload: {:?}", e);

                    // Commit the message offset.
                    if let Err(e) = consumer_arc.commit_message(&borrowed_message, rdkafka::consumer::CommitMode::Async) {
                        println!("Error committing offset: {:?}", e);
                    }

                    // Release the semaphore permit so that we can continue processing messages.
                    drop(semaphore_permit); 

                    continue;
                }
            }.to_bytes().to_vec();

            // Detach the message from the consumer so that we can process it in a separate thread.
            let owned_message = borrowed_message.detach();
            
            tokio::spawn(async move {
                
                println!("Received message: {:?}", payload);
                
                // It simulates sending to another microservice on which I must take care not to generate backpressure.
                tokio::time::sleep(Duration::from_secs(1)).await;

                /*  Commit the message offset within an asynchronous task
                
                Is it possible to commit the messages that were processed within the asynchronous task and leave uncommitted those that 
                were not taken by the asynchronous task?
                
                let mut tpl = rdkafka::TopicPartitionList::new();
                
                let _ = tpl.add_partition_offset(
                    owned_message.topic(),
                    owned_message.partition(),
                    rdkafka::Offset::Offset(owned_message.offset()),
                );

                if let Err(e) = consumer_clone.commit(&tpl, rdkafka::consumer::CommitMode::Async)
                {
                    println!("Error committing offsets: {:?}", e);
                }
                */
                
                // Release the semaphore permit so that we can continue processing messages.
                drop(semaphore_permit);

            });
        }
    }
}

rdkafka = { version = "0.35.0", features = ["cmake-build", "ssl-vendored"] }

[dertin]

Hi, @benesch
#226 (comment) this doesn't work right?