feat: add BrokerPool #4
Conversation
src/client.rs
Outdated
| for upper_bound in (ApiVersionsRequest::API_VERSION_RANGE.0 .0 .0 | ||
| ..=ApiVersionsRequest::API_VERSION_RANGE.1 .0 .0) | ||
| .rev() | ||
| I: IntoIterator<Item = A>, |
There was a problem hiding this comment.
Is it really necessary to take an iterator here? Nobody is going to pass in thousand of connections here, so I think a simple vector would also do the job and would result in a somewhat more ergonomic interface.
tests/client.rs
Outdated
| @@ -49,5 +49,5 @@ macro_rules! maybe_skip_kafka_integration { | |||
| #[tokio::test] | |||
| async fn test_connect() { | |||
| let connection = maybe_skip_kafka_integration!(); | |||
| Client::new(connection).await; | |||
| Client::new(std::iter::once(connection)).await; | |||
There was a problem hiding this comment.
Even when we keep the IntoIter type, I think vec![connection] is easier to read.
src/messenger.rs
Outdated
| /// A connection to a single broker | ||
| /// | ||
| /// Note: Requests to the same [`Messenger`] will be pipelined by Kafka | ||
| /// - |
There was a problem hiding this comment.
| /// - |
Or was there more?
| ); | ||
| Duration::from_secs_f64(std::mem::replace(&mut self.next_backoff_secs, next_backoff)) | ||
| } | ||
| } |
There was a problem hiding this comment.
| } | |
| } | |
| // TODO: tests (either statistical or by using a fixed RNG) |
| let brokers = if self.discovered_brokers.is_empty() { | ||
| &self.bootstrap_brokers | ||
| } else { | ||
| &self.discovered_brokers | ||
| }; |
There was a problem hiding this comment.
How about chaining this these two lists (discovered first, then falling back to bootstrap) in case the cluster got into some weird state (e.g. you have two brokers, the bootstrap broker doesn't report itself and after a while comes back and the discovered broker dies).
There was a problem hiding this comment.
The discovered brokers should include the bootstrap brokers - will add docs to clarify
tustvold
added
automerge
This adds a
BrokerPoolthat tries to connect to one of a list of bootstrap brokers, and then caches this connection.A future PR will add functionality to fetch the full list of brokers and also cache this.
This is following the docs here