kafka message queue backend is not working with my code

[neerajdlh74]

I needed to use kafka as backend message queue, but it is not working, and also it is not throwing any error just connection is completed but it is stuck on websocket protocol upgrade, i checked that kafka topics and brokers are working properly, but don't know where the problem could be. I used gevent for running the server. please someone help me.

here i am pasting server.py code:

import socketio
from gevent import pywsgi
from geventwebsocket.handler import WebSocketHandler


url=['kafka://127.0.0.1:9092', 'kafka://127.0.0.2:9092', 'kafka://127.0.0.3:9092']
mgr = socketio.KafkaManager(url=url, channel='sockets')
sio = socketio.Server(async_mode='gevent', client_manager=mgr, logger=True, engineio_logger=True)
app = socketio.WSGIApp(sio)

@sio.event
def connect(sid, environ):
      print('connected')
      sio.emit('message', {'foo':'bar'})

@sio.event
def message(sid, data):
      print('msg', data)

@sio.event
def disconnect(sid):
      print('disconnect')

if __name__=='__main__':
    pywsgi.WSGIServer(('0.0.0.0', 8000), app, handler_class=WebSocketHandler).serve_forever()

I tried using monkey patch but i got this error:

Traceback (most recent call last):
  File "server.py", line 12, in <module>
    mgr = socketio.KafkaManager(url=url, channel='sockets')
  File "/usr/local/lib/python3.6/site-packages/socketio/kafka_manager.py", line 52, in __init__
    self.producer = kafka.KafkaProducer(bootstrap_servers=self.kafka_urls)
  File "/usr/local/lib/python3.6/site-packages/kafka/producer/kafka.py", line 383, in __init__
    **self.config)
  File "/usr/local/lib/python3.6/site-packages/kafka/client_async.py", line 208, in __init__
    self._selector = self.config['selector']()
  File "/usr/lib64/python3.6/selectors.py", line 399, in __init__
    self._epoll = select.epoll()
AttributeError: module 'select' has no attribute 'epoll'
Exception ignored in: <bound method KafkaClient.__del__ of <kafka.client_async.KafkaClient object at 0x7fc910692358>>
Traceback (most recent call last):
  File "/usr/local/lib/python3.6/site-packages/kafka/client_async.py", line 443, in __del__
    self._close()
  File "/usr/local/lib/python3.6/site-packages/kafka/client_async.py", line 421, in _close
    self._selector.close()
AttributeError: 'KafkaClient' object has no attribute '_selector'

here is the client.py code for connecting to the server:

import asyncio
import socketio


class Check:
    def __init__(self):
        self.sio = socketio.AsyncClient(handle_sigint=False, logger=True, engineio_logger=True)
        self.sio.on('connect', self.connect, namespace='/')
        self.sio.on('message', self.message, namespace='/')
        self.sio.on('connect_error', self.connect_error, namespace='/')
        self.sio.on('disconnect', self.disconnect, namespace='/')



    async def main_connect(self):
        username = generate_username(1)[0]
        await self.sio.connect('http://0.0.0.0:8000', wait_timeout = 10)


    async def connect(self):
        print('connected')
        await self.sio.emit('message', 'hello world')


    async def connect_error(self,e):
        print(e)
    async def message(self, data):
        print('msg', data)
    async def disconnect(self):
        print('disconnected')

    async def run(self):
        await self.main_connect()

async def main():
    async def loop(i):
        await asyncio.sleep(i/10)
        obj1 = Check()
        await obj1.run()
        await obj1.sio.wait()
    x = [loop(i) for i in range(1)]
    await asyncio.gather(*x)

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(main())

[miguelgrinberg]

Unfortunately this isn't something I can help with. You are having incompatibility issues between the Kafka library and gevent. Without monkey patching I'm fairly sure nothing will work. If this library fails after gevent monkey patching then the only other option I can do think is to try eventlet instead of gevent, also with monkey patching.

If none of that works, then you won't be able to use gevent or eventlet. The threading async mode will work fine, if you can manage its limitations.

[neerajdlh74]

I tried using eventlet, without monkey patch it worked, but with monkey patching it gave same error as gevent, after that i added this parameter in monkey patching eventlet.monkey_patch(socket=True, select=False) and it worked,
i tried repeating this with gevent but it didn't worked there.

my requirement is to use kafka with gevent only
if you could help me find correct compatible library so that kafka should work with gevent, it will be much helpful.

Here is the version details currently I am using right now:

Python 3.6.8

gevent==1.2.2
gevent-websocket==0.10.1
eventlet==0.33.1
kafka-python==2.0.2
python-engineio==4.3.2
python-socketio==5.6.0
websocket-client==1.3.1
websockets==9.1
greenlet==1.1.2
asyncio==3.4.3

[miguelgrinberg]

I'm sorry, but this is not an area that I can help. This is an incompatibility between gevent and kafka-python. You will need to ask the people associated with these packages to see if they have any suggestions.