def _send_broker_aware_request(self, payloads, encoder_fn, decoder_fn):
        """
        Group a list of request payloads by topic+partition and send them to
        the leader broker for that partition using the supplied encode/decode
        functions

        Arguments:

        payloads: list of object-like entities with a topic (str) and
            partition (int) attribute; payloads with duplicate topic-partitions
            are not supported.

        encode_fn: a method to encode the list of payloads to a request body,
            must accept client_id, correlation_id, and payloads as
            keyword arguments

        decode_fn: a method to decode a response body into response objects.
            The response objects must be object-like and have topic
            and partition attributes

        Returns:

        List of response objects in the same order as the supplied payloads
        """
        # encoders / decoders do not maintain ordering currently
        # so we need to keep this so we can rebuild order before returning
        original_ordering = [(p.topic, p.partition) for p in payloads]

        # Group the requests by topic+partition
        brokers_for_payloads = []
        payloads_by_broker = collections.defaultdict(list)

        responses = {}
        for payload in payloads:
            try:
                leader = self._get_leader_for_partition(payload.topic,
                                                        payload.partition)
                payloads_by_broker[leader].append(payload)
                brokers_for_payloads.append(leader)
            except KafkaUnavailableError as e:
                log.warning('KafkaUnavailableError attempting to send request '
                            'on topic %s partition %d', payload.topic, payload.partition)
                topic_partition = (payload.topic, payload.partition)
                responses[topic_partition] = FailedPayloadsError(payload)

        # For each broker, send the list of request payloads
        # and collect the responses and errors
        broker_failures = []

        # For each KafkaConnection keep the real socket so that we can use
        # a select to perform unblocking I/O
        connections_by_socket = {}
        for broker, payloads in payloads_by_broker.items():
            requestId = self._next_id()
            log.debug('Request %s to %s: %s', requestId, broker, payloads)
            request = encoder_fn(client_id=self.client_id,
                                 correlation_id=requestId, payloads=payloads)

            # Send the request, recv the response
            try:
                conn = self._get_conn(broker.host.decode('utf-8'), broker.port)
                conn.send(requestId, request)

            except ConnectionError as e:
                broker_failures.append(broker)
                log.warning('ConnectionError attempting to send request %s '
                            'to server %s: %s', requestId, broker, e)

                for payload in payloads:
                    topic_partition = (payload.topic, payload.partition)
                    responses[topic_partition] = FailedPayloadsError(payload)

            # No exception, try to get response
            else:

                # decoder_fn=None signal that the server is expected to not
                # send a response.  This probably only applies to
                # ProduceRequest w/ acks = 0
                if decoder_fn is None:
                    log.debug('Request %s does not expect a response '
                              '(skipping conn.recv)', requestId)
                    for payload in payloads:
                        topic_partition = (payload.topic, payload.partition)
                        responses[topic_partition] = None
                    continue
                else:
                    connections_by_socket[conn.get_connected_socket()] = (conn, broker, requestId)

        conn = None
        while connections_by_socket:
            sockets = connections_by_socket.keys()
            rlist, _, _ = select.select(sockets, [], [], None)
            conn, broker, requestId = connections_by_socket.pop(rlist[0])
            try:
                response = conn.recv(requestId)
            except ConnectionError as e:
                broker_failures.append(broker)
                log.warning('ConnectionError attempting to receive a '
                            'response to request %s from server %s: %s',
                            requestId, broker, e)

                for payload in payloads_by_broker[broker]:
                    topic_partition = (payload.topic, payload.partition)
                    responses[topic_partition] = FailedPayloadsError(payload)

            else:
                _resps = []
                for payload_response in decoder_fn(response):
                    topic_partition = (payload_response.topic,
                                       payload_response.partition)
                    responses[topic_partition] = payload_response
                    _resps.append(payload_response)
                log.debug('Response %s: %s', requestId, _resps)

        # Connection errors generally mean stale metadata
        # although sometimes it means incorrect api request
        # Unfortunately there is no good way to tell the difference
        # so we'll just reset metadata on all errors to be safe
        if broker_failures:
            self.reset_all_metadata()

        # Return responses in the same order as provided
        return [responses[tp] for tp in original_ordering]
def _send_broker_aware_request ( self , payloads , encoder_fn , decoder_fn ) : original_ordering = [ ( p . topic , p . partition ) for p in payloads ] brokers_for_payloads = [ ] payloads_by_broker = collections . defaultdict ( list ) responses = { } for payload in payloads : try : leader = self . _get_leader_for_partition ( payload . topic , payload . partition ) payloads_by_broker [ leader ] . append ( payload ) brokers_for_payloads . append ( leader ) except KafkaUnavailableError as e : log . warning ( 'KafkaUnavailableError attempting to send request ' 'on topic %s partition %d' , payload . topic , payload . partition ) topic_partition = ( payload . topic , payload . partition ) responses [ topic_partition ] = FailedPayloadsError ( payload ) broker_failures = [ ] connections_by_socket = { } for broker , payloads in payloads_by_broker . items ( ) : requestId = self . _next_id ( ) log . debug ( 'Request %s to %s: %s' , requestId , broker , payloads ) request = encoder_fn ( client_id = self . client_id , correlation_id = requestId , payloads = payloads ) try : conn = self . _get_conn ( broker . host . decode ( 'utf-8' ) , broker . port ) conn . send ( requestId , request ) except ConnectionError as e : broker_failures . append ( broker ) log . warning ( 'ConnectionError attempting to send request %s ' 'to server %s: %s' , requestId , broker , e ) for payload in payloads : topic_partition = ( payload . topic , payload . partition ) responses [ topic_partition ] = FailedPayloadsError ( payload ) else : if decoder_fn is None : log . debug ( 'Request %s does not expect a response ' '(skipping conn.recv)' , requestId ) for payload in payloads : topic_partition = ( payload . topic , payload . partition ) responses [ topic_partition ] = None continue else : connections_by_socket [ conn . get_connected_socket ( ) ] = ( conn , broker , requestId ) conn = None while connections_by_socket : sockets = connections_by_socket . keys ( ) rlist , _ , _ = select . select ( sockets , [ ] , [ ] , None ) conn , broker , requestId = connections_by_socket . pop ( rlist [ 0 ] ) try : response = conn . recv ( requestId ) except ConnectionError as e : broker_failures . append ( broker ) log . warning ( 'ConnectionError attempting to receive a ' 'response to request %s from server %s: %s' , requestId , broker , e ) for payload in payloads_by_broker [ broker ] : topic_partition = ( payload . topic , payload . partition ) responses [ topic_partition ] = FailedPayloadsError ( payload ) else : _resps = [ ] for payload_response in decoder_fn ( response ) : topic_partition = ( payload_response . topic , payload_response . partition ) responses [ topic_partition ] = payload_response _resps . append ( payload_response ) log . debug ( 'Response %s: %s' , requestId , _resps ) if broker_failures : self . reset_all_metadata ( ) return [ responses [ tp ] for tp in original_ordering ]
Group a list of request payloads by topic+partition and send them to
        the leader broker for that partition using the supplied encode/decode
        functions

        Arguments:

        payloads: list of object-like entities with a topic (str) and
            partition (int) attribute; payloads with duplicate topic-partitions
            are not supported.

        encode_fn: a method to encode the list of payloads to a request body,
            must accept client_id, correlation_id, and payloads as
            keyword arguments

        decode_fn: a method to decode a response body into response objects.
            The response objects must be object-like and have topic
            and partition attributes

        Returns:

        List of response objects in the same order as the supplied payloads