def add_coordinator(self, response, coord_type, coord_key):
        """Update with metadata for a group or txn coordinator

        Arguments:
            response (FindCoordinatorResponse): broker response
            coord_type (str): 'group' or 'transaction'
            coord_key (str): consumer_group or transactional_id

        Returns:
            string: coordinator node_id if metadata is updated, None on error
        """
        log.debug("Updating coordinator for %s/%s: %s", coord_type, coord_key, response)
        error_type = Errors.for_code(response.error_code)
        if error_type is not Errors.NoError:
            log.error("FindCoordinatorResponse error: %s", error_type)
            self._coordinators[(coord_type, coord_key)] = -1
            return

        # Use a coordinator-specific node id so that requests
        # get a dedicated connection
        node_id = 'coordinator-{}'.format(response.coordinator_id)
        coordinator = BrokerMetadata(
            node_id,
            response.host,
            response.port,
            None)

        log.info("Coordinator for %s/%s is %s", coord_type, coord_key, coordinator)
        self._coordinator_brokers[node_id] = coordinator
        self._coordinators[(coord_type, coord_key)] = node_id
        return node_id
Update with metadata for a group or txn coordinator

        Arguments:
            response (FindCoordinatorResponse): broker response
            coord_type (str): 'group' or 'transaction'
            coord_key (str): consumer_group or transactional_id

        Returns:
            string: coordinator node_id if metadata is updated, None on error
        
kafka/cluster.py