private void handleCompletedReceives(List<ClientResponse> responses, long now) {
        for (NetworkReceive receive : this.selector.completedReceives()) {
            String source = receive.source();
            InFlightRequest req = inFlightRequests.completeNext(source);

            AbstractResponse response = parseResponse(receive.payload(), req.header);
            if (throttleTimeSensor != null)
                throttleTimeSensor.record(response.throttleTimeMs(), now);

            if (log.isDebugEnabled()) {
                log.debug("Received {} response from node {} for request with header {}: {}",
                    req.header.apiKey(), req.destination, req.header, response);
            }

            // If the received response includes a throttle delay, throttle the connection.
            maybeThrottle(response, req.header.apiVersion(), req.destination, now);
            if (req.isInternalRequest && response instanceof MetadataResponse)
                metadataUpdater.handleSuccessfulResponse(req.header, now, (MetadataResponse) response);
            else if (req.isInternalRequest && response instanceof ApiVersionsResponse)
                handleApiVersionsResponse(responses, req, now, (ApiVersionsResponse) response);
            else if (req.isInternalRequest && response instanceof GetTelemetrySubscriptionsResponse)
                telemetrySender.handleResponse((GetTelemetrySubscriptionsResponse) response);
            else if (req.isInternalRequest && response instanceof PushTelemetryResponse)
                telemetrySender.handleResponse((PushTelemetryResponse) response);
            else
                responses.add(req.completed(response, now));
        }
    }
Handle any completed receives and update the response list with the responses received.

@param responses The list of responses to update
@param now The current time
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java