feat(mqtt): Add single shared client per mqtt broker for split enumerator #23003
Conversation
There was a problem hiding this comment.
Pull Request Overview
This PR refactors the MQTT split enumerator to use a single shared client per MQTT broker instead of creating individual clients for each source, resolving client ID conflicts and improving resource efficiency.
Key Changes
- Introduced a shared client cache using
MokaCacheto store weak references to connection objects per broker - Extracted connection management logic into a dedicated
MqttConnectionCheckstruct - Modified the enumerator to reuse existing connections when available or create new ones as needed
| connected: Arc<AtomicBool>, | ||
| stopped: Arc<AtomicBool>, |
There was a problem hiding this comment.
Is it possible to remove state for stopped? seems connected is enough.
There was a problem hiding this comment.
Stopped is used to stop the loop running in the background. (see line 65)
When MqttConnectionCheck is dropped(see line 91), this is set to false, which stops the loop.
Happy to hear any other way in which this can be achieved.
| let start = std::time::Instant::now(); | ||
| loop { | ||
| if self.connected.load(std::sync::atomic::Ordering::Relaxed) { | ||
| if self.connection_check.is_connected() { |
There was a problem hiding this comment.
shall we add some tokio sleep inside the loop?
There was a problem hiding this comment.
There is a 500 ms sleep on line 154 that gets executed when the client is not connected.
When the client is connected the entire loop stops, and is not started again until the next time list_splits is called.
I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.
What's changed and what's your intention?
The MQTTSplitEnumerator used to make a client for each source. We had some issues with that since it uses the same client id. The better approach here is to use a shared client for each broker, as the done with the kafka connector
Checklist
Documentation
Release note