Assertion error in aiokafka sender routine causing app to stall

[vineetgoel]

The app seems to stall after the following errors. We should just make the app crash instead of stalling so that it is restarted by supervisor:

[2018-05-29 14:08:03,304: ERROR]: Future exception was never retrieved
future: <Future finished exception=NodeNotReadyError('Attempt to send a request to node which is not ready (node id 42).',)>
Traceback (most recent call last):
  File "/home/robinhood/env/lib/python3.6/site-packages/aiokafka/producer/producer.py", line 419, in _send_produce_req
    response = yield from self.client.send(node_id, request)
  File "/home/robinhood/env/lib/python3.6/site-packages/aiokafka/client.py", line 406, in send
    " which is not ready (node id {}).".format(node_id))
kafka.errors.NodeNotReadyError: NodeNotReadyError: Attempt to send a request to node which is not ready (node id 42).
[2018-05-29 14:08:03,315: ERROR]: Unable connect to node with id 42: [Errno 111] Connect call failed ('10.1.30.204', 9092)
[2018-05-29 14:08:03,315: WARNING]: Got error produce response: NodeNotReadyError: Attempt to send a request to node which is not ready (node id 42).
[2018-05-29 14:08:03,715: ERROR]: Unable connect to node with id 42: [Errno 111] Connect call failed ('10.1.30.204', 9092)
[2018-05-29 14:08:03,716: WARNING]: Got error produce response: NodeNotReadyError: Attempt to send a request to node which is not ready (node id 42).
[2018-05-29 14:08:04,117: ERROR]: Unable connect to node with id 42: [Errno 111] Connect call failed ('10.1.30.204', 9092)
[2018-05-29 14:08:04,117: WARNING]: Got error produce response: NodeNotReadyError: Attempt to send a request to node which is not ready (node id 42).
[2018-05-29 14:08:04,517: ERROR]: Unable connect to node with id 42: [Errno 111] Connect call failed ('10.1.30.204', 9092)
[2018-05-29 14:08:04,518: WARNING]: Got error produce response: NodeNotReadyError: Attempt to send a request to node which is not ready (node id 42).
[2018-05-29 14:08:04,917: ERROR]: Unexpected error in sender routine
Traceback (most recent call last):
  File "/home/robinhood/env/lib/python3.6/site-packages/aiokafka/producer/producer.py", line 374, in _sender_routine
    task.result()
  File "/home/robinhood/env/lib/python3.6/site-packages/aiokafka/producer/producer.py", line 419, in _send_produce_req
    response = yield from self.client.send(node_id, request)
  File "/home/robinhood/env/lib/python3.6/site-packages/aiokafka/client.py", line 403, in send
    if not (yield from self.ready(node_id, group=group)):
  File "/home/robinhood/env/lib/python3.6/site-packages/aiokafka/client.py", line 381, in ready
    conn = yield from self._get_conn(node_id, group=group)
  File "/home/robinhood/env/lib/python3.6/site-packages/aiokafka/client.py", line 354, in _get_conn
    assert broker, 'Broker id %s not in current metadata' % node_id
AssertionError: Broker id 42 not in current metadata