code: def _connect(self):
          
          def tryConnect():
              self.connector = d = maybeDeferred(connect)
              d.addCallback(cbConnect)
              d.addErrback(ebConnect)
  
          def connect():
              endpoint = self._endpointFactory(self._reactor, self.host, self.port)
              log.debug('%r: connecting with %s', self, endpoint)
              return endpoint.connect(self)
  
          def cbConnect(proto):
              log.debug('%r: connected to %r', self, proto.transport.getPeer())
              self._failures = 0
              self.connector = None
              self.proto = proto
              if self._dDown:
                  proto.transport.loseConnection()
              else:
                  self._sendQueued()
  
          def ebConnect(fail):
              if self._dDown:
                  log.debug('%r: breaking connect loop due to %r after close()', self, fail)
                  return fail
              self._failures += 1
              delay = self._retryPolicy(self._failures)
              log.debug('%r: failure %d to connect -> %s; retry in %.2f seconds.',
                        self, self._failures, fail.value, delay)
  
              self.connector = d = deferLater(self._reactor, delay, lambda: None)
              d.addCallback(cbDelayed)
  
          def cbDelayed(result):
              tryConnect()
  
          self._failures = 0
          tryConnect()
docs: Connect to the Kafka Broker
  
          This routine will repeatedly try to connect to the broker (with backoff
          according to the retry policy) until it succeeds.