StreamConsumer stuck after commit, next consume call never returns (worse after Kafka 4.2.0 upgrade)
Hi,
We are using rdkafka crate version 0.36.2 with StreamConsumer (tokio runtime).
We have a common library where all our Kafka consumer/producer code is written, and all our projects use this library. Consumers are created once and stored in a lazy_static map so we can reuse the same StreamConsumer instance:
lazy_static! {
pub static ref CONSUMERS: RwLock<HashMap<String, HashMap<String, Arc<StreamConsumer<ConsumerCallbackLogger>>>>> =
RwLock::new(HashMap::new());
}Our consumer config:
enable.auto.commit = false
max.poll.interval.ms = 300000
fetch.wait.max.ms = 500
enable.partition.eof = true
statistics.interval.ms = 30000
security.protocol
sasl.mechanism
ssl.ca.location
ssl.endpoint.identification.algorithm
enable.ssl.certificate.verification
We consume messages in batches like this:
Our consume function:
pub async fn consume_msg(&self, batch_size: i64) -> Result<Vec<BorrowedMessage>, String> {
let mut messages = Vec::with_capacity(batch_size as usize);
let mut return_vec = Vec::with_capacity(batch_size as usize);
while messages.len() < messages.capacity() {
let message_stream = self.consumer.stream().next().await;
match message_stream {
Some(result) => match result {
Ok(msg) => messages.push(msg),
Err(err) => {
if let KafkaError::PartitionEOF(_) = err {
if !messages.is_empty() {
return_vec.append(&mut messages);
return Ok(return_vec);
}
}
if let Some(code) = err.rdkafka_error_code() {
self.handle_consumer_error(code);
}
return Err(err.to_string());
}
},
None => {
if !messages.is_empty() {
return_vec.append(&mut messages);
return Ok(return_vec);
}
return Err("No messages received from Kafka".to_string());
}
}
}
return_vec.append(&mut messages);
Ok(return_vec)
}The problem we are facing:
Sometimes processing of one batch takes 2-3 minutes. After processing, we commit offsets manually and call consume again. Normally, if the topic is idle, .next().await simply waits for new messages — that behavior is expected and we are aware of it.
The problem is when the errors mentioned below occur: even though there are still messages pending in the topic (consumer lag exists), .next().await returns nothing — no message, no error, no None. It seems the library does not surface this error through the stream, so we have no way to detect or handle it in our code.
Once this happens, the consumer never receives messages again until we restart the service. It doesn't happen every time — around 90% of the time it works fine — but the frequency has increased a lot in the last week, after our Kafka brokers were upgraded to 4.0.2.
When this happens we see these errors in logs:
%3|1784271798.128|FAIL|rdkafka#producer-481| [thrd:sasl_ssl://data-kafka-bootstrap:9095/bootstrap]: sasl_ssl://data-kafka-bootstrap:9095/bootstrap: Disconnected: SSL connection closed by peer (after 0ms in state SSL_HANDSHAKE)
ERROR - librdkafka: Global error: BrokerTransportFailure (Local: Broker transport failure): sasl_ssl://data-kafka-bootstrap:9095/bootstrap: SSL handshake failed: Disconnected: connecting to a PLAINTEXT broker listener? (after 9997ms in state SSL_HANDSHAKE, 3 identical error(s) suppressed)
These transport errors only come in the global error callback, they are never returned as Err from the stream, so our code has no way to detect it — the stream just goes silent.
Can you please help us understand why the consumer gets stuck like this after a broker disconnect, and what is the correct way to detect and recover from this situation with StreamConsumer? Also is it a problem that we call consumer.stream().next().await in a loop (new stream per message) instead of using recv()?
server kafka version: 4.2.0
librdkafka version: 1.36.2
Rust version: 1.89
Thanks.
StreamConsumer stuck after commit, next consume call never returns (worse after Kafka 4.2.0 upgrade)
Hi,
We are using rdkafka crate version 0.36.2 with StreamConsumer (tokio runtime).
We have a common library where all our Kafka consumer/producer code is written, and all our projects use this library. Consumers are created once and stored in a lazy_static map so we can reuse the same StreamConsumer instance:
Our consumer config:
We consume messages in batches like this:
Our consume function:
The problem we are facing:
Sometimes processing of one batch takes 2-3 minutes. After processing, we commit offsets manually and call consume again. Normally, if the topic is idle, .next().await simply waits for new messages — that behavior is expected and we are aware of it.
The problem is when the errors mentioned below occur: even though there are still messages pending in the topic (consumer lag exists), .next().await returns nothing — no message, no error, no None. It seems the library does not surface this error through the stream, so we have no way to detect or handle it in our code.
Once this happens, the consumer never receives messages again until we restart the service. It doesn't happen every time — around 90% of the time it works fine — but the frequency has increased a lot in the last week, after our Kafka brokers were upgraded to 4.0.2.
When this happens we see these errors in logs:
These transport errors only come in the global error callback, they are never returned as Err from the stream, so our code has no way to detect it — the stream just goes silent.
Can you please help us understand why the consumer gets stuck like this after a broker disconnect, and what is the correct way to detect and recover from this situation with StreamConsumer? Also is it a problem that we call
consumer.stream().next().awaitin a loop (new stream per message) instead of usingrecv()?server kafka version: 4.2.0
librdkafka version: 1.36.2
Rust version: 1.89
Thanks.