title: How to handle connection issues with kafka using the python kafka library?
body: <p>I have a serverless function that's trying to send some data to kafka. Sometimes it works and sometimes the connection just drop and the data is lost.</p>
  
  <p>The reason for this is that the library for kafka is not raising the exception but it's adding error logs instead. So i cannot add my piece of code in a <code>try:except</code>.</p>
  
  <p>Here is the error that i am often getting in my logs:</p>
  
  <pre><code>&lt;BrokerConnection node_id=11 host=... port=9092&gt;: Error receiving network data closing socket
  Traceback (most recent call last):
  File "/var/task/kafka/conn.py", line 745, in _recv
  data = self._sock.recv(SOCK_CHUNK_BYTES)
  ConnectionResetError: [Errno 104] Connection reset by peer
  </code></pre>
  
  <p>and the function <code>_recv</code> mentioned above has definition as follows:</p>
  
  <p>I am still looking for a solution but adding the code within try:except doesn't work.</p>
  
  <pre><code>def _recv(self):
          responses = []
          SOCK_CHUNK_BYTES = 4096
          while True:
              try:
                  data = self._sock.recv(SOCK_CHUNK_BYTES)
                  # We expect socket.recv to raise an exception if there is not
                  # enough data to read the full bytes_to_read
                  # but if the socket is disconnected, we will get empty data
                  # without an exception raised
                  if not data:
                      log.error('%s: socket disconnected', self)
                      self.close(error=Errors.ConnectionError('socket disconnected'))
                      break
                  else:
                      responses.extend(self.receive_bytes(data))
                      if len(data) &lt; SOCK_CHUNK_BYTES:
                          break
              except SSLWantReadError:
                  break
              except ConnectionError as e:
                  if six.PY2 and e.errno == errno.EWOULDBLOCK:
                      break
                  log.exception('%s: Error receiving network data'
                                ' closing socket', self)
                  self.close(error=Errors.ConnectionError(e))
                  break
              except BlockingIOError:
                  if six.PY3:
                      break
                  raise
          return responses
  </code></pre>
  
  <p>I expect when the connection is lost, an exception is raised and the application breaks but instead only there are only error logs and the application continues even if the kafka part fails.</p>