From d6ec6e15963fe263f3dcefbfbbb0f4214fff3f06 Mon Sep 17 00:00:00 2001 From: Samuel Cantero Date: Thu, 4 Jan 2024 16:01:22 -0300 Subject: [PATCH] Return back to the caller on rebalance events This does not affect the StreamConsumer or any other wrapper consumer. It will only incur on an extra Poll call when there's a rebalance event. When using bindings built upon the rust-rdkafka ffi, the caller is responsible for initiating the rebalance calls (*assign). If a high timeout is specified, the rebalance handler will only be triggered once the timeout period has elapsed. This fixes it by always returning on rebalance events except when Timeout::Never. Poll calls with timeout::Never are expected to return a message. --- src/consumer/base_consumer.rs | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/src/consumer/base_consumer.rs b/src/consumer/base_consumer.rs index 08ec51b78..2a45306c5 100644 --- a/src/consumer/base_consumer.rs +++ b/src/consumer/base_consumer.rs @@ -144,9 +144,15 @@ where } rdsys::RD_KAFKA_EVENT_REBALANCE => { self.handle_rebalance_event(event); + if timeout != Timeout::Never { + return None; + } } rdsys::RD_KAFKA_EVENT_OFFSET_COMMIT => { self.handle_offset_commit_event(event); + if timeout != Timeout::Never { + return None; + } } _ => { let buf = unsafe {