Skip to content

Commit

Permalink
Return back to the caller on rebalance events
Browse files Browse the repository at this point in the history
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.
  • Loading branch information
scanterog authored and davidblewett committed Jan 11, 2024
1 parent 738590b commit d6ec6e1
Showing 1 changed file with 6 additions and 0 deletions.
6 changes: 6 additions & 0 deletions src/consumer/base_consumer.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down

0 comments on commit d6ec6e1

Please sign in to comment.