How to use multiple kafka server in python-socketio?

[sparkingdark]

I have a doubt regarding the python flask socketio that let's say I have a multi broker single zookeper kafka docker container running with address

• broker 1 : localhost:9092
• broker 2 : localhost:9093

now in kafka offical client I can create prodiucer and consumer

my_producer = KafkaProducer(  
    bootstrap_servers = ['localhost:9092', 'localhost:9093', 'localhost:9094'],  
    value_serializer = lambda x:dumps(x).encode('utf-8')  
    ) 

which will use all the server in case a load scenerio present. But in python sockerio I can only connect to a single instance of kafka broker so How can I pass all the multiple broker ip into the kafkamanager.

mgr = socketio.KafkaManager("kafka://" + KAFKA_IP_PORT, write_only=False)

How can I pass like

mgr = socketio.KafkaManager(server_urls=[broker1 ip,broker2 ip], write_only=False)

[miguelgrinberg]

The Kafka client code was not written by myself, it was contributed by a user. The option to pass multiple servers was not implemented by this user, so this is currently not possible. Would you be interested in extending the logic to optionally allow a list of URLs to be provided?

[sparkingdark]

Hello @miguelgrinberg can you check this logic I added into the https://github.com/sparkingdark/python-socketio/blob/dacdc1a13a794d073b220dfa3440e5556ace3e3e/src/socketio/kafka_manager.py#L38

here is correct way to start to implement the feature currently I will do that rest of testing and implementation this weekend.

[miguelgrinberg]

To avoid a long list of arguments, my suggestion is that you use the same url argument for this. If a string is provided, then it is a single server. If a list is provided, then you treat each of the elements in the list as connection URLs. Would that work?

[sparkingdark]

okay got the point there would be a check like this:

if type(url) == List:
   pass kafkamanager(bootstrap_server=url)

[sparkingdark]

@miguelgrinberg is there any slack or discord community channel available for python socketio and flaskio

[miguelgrinberg]

Try something like this, so that you don't have to have separate logic for connecting one or multiple servers, and also so that you apply the same pre-processing to the list than it is currently applied to a single URL:

        urls = [url] if isinstance(url, str) else url
        self.kafka_urls = [url[8:] if url != 'kafka://' else 'localhost:9092' for url in urls]
        self.producer = kafka.KafkaProducer(bootstrap_servers=self.kafka_urls)
        self.consumer = kafka.KafkaConsumer(self.channel,
                                            bootstrap_servers=self.kafka_urls)

And no, there is no live chat channel at this time. I don't have the bandwidth to host one (have too many projects going on).

Also, type(x) is almost always not a good idea as this checks for the exact type. isinstance(x, type) is better as it also works with subclasses of the type being checked.

[sparkingdark]

Thanks for your reply. I will do that and let you know by pr. A live channel would be awesome but GitHub discussion is quite good.

…

On Fri, 24 Dec, 2021, 10:07 pm Miguel Grinberg, ***@***.***> wrote: Try something like this, so that you don't have to have separate logic for connecting one or multiple servers, and also so that you apply the same pre-processing to the list than it is currently applied to a single URL: urls = [url] if isinstance(url, str) else url self.kafka_urls = [url[8:] if url != 'kafka://' else 'localhost:9092' for url in urls] self.producer = kafka.KafkaProducer(bootstrap_servers=self.kafka_urls) self.consumer = kafka.KafkaConsumer(self.channel, bootstrap_servers=self.kafka_urls) And no, there is no live chat channel at this time. I don't have the bandwidth to host one (have too many projects going on). — Reply to this email directly, view it on GitHub <#1753 (reply in thread)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AL7XHCESLVO6IJYFL5J4YPTUSSOSZANCNFSM5KR2CYCQ> . You are receiving this because you authored the thread.Message ID: <miguelgrinberg/Flask-SocketIO/repo-discussions/1753/comments/1867967@ github.com>