code: def _close_brokerclients(self, clients):
          
          def _log_close_failure(failure, brokerclient):
              log.debug(
                  'BrokerClient: %s close result: %s: %s', brokerclient,
                  failure.type.__name__, failure.getErrorMessage())
  
          def _clean_close_dlist(result, close_dlist):
              # If there aren't any other outstanding closings going on, then
              # close_dlist == self.close_dlist, and we can reset it.
              if close_dlist == self.close_dlist:
                  self.close_dlist = None
  
          if not self.close_dlist:
              dList = []
          else:
              log.debug("%r: _close_brokerclients has nested deferredlist: %r",
                        self, self.close_dlist)
              dList = [self.close_dlist]
          for brokerClient in clients:
              log.debug("Calling close on: %r", brokerClient)
              d = brokerClient.close().addErrback(_log_close_failure, brokerClient)
              dList.append(d)
          self.close_dlist = DeferredList(dList)
          self.close_dlist.addBoth(_clean_close_dlist, self.close_dlist)
docs: Close the given broker clients.
  
          :param clients: Iterable of `_KafkaBrokerClient`