docs: def reinit(self):
          """
          Re-initialize the socket connection
          close current socket (if open)
          and start a fresh connection
          raise ConnectionError on error
          """
          log.debug("Reinitializing socket connection for %s:%d" % (self.host, self.port))
  
          if self._sock:
              self.close()
  
          try:
              self._sock = socket.create_connection((self.host, self.port), self.timeout)
          except socket.error:
              log.exception('Unable to connect to kafka broker at %s:%d' % (self.host, self.port))
              self._raise_connection_error()