public static ClientResponse sendAndReceive(KafkaClient client, ClientRequest request, Time time) throws IOException {
        try {
            client.send(request, time.milliseconds());
            while (client.active()) {
                List<ClientResponse> responses = client.poll(Long.MAX_VALUE, time.milliseconds());
                for (ClientResponse response : responses) {
                    if (response.requestHeader().correlationId() == request.correlationId()) {
                        if (response.wasDisconnected()) {
                            throw new IOException("Connection to " + response.destination() + " was disconnected before the response was read");
                        }
                        if (response.versionMismatch() != null) {
                            throw response.versionMismatch();
                        }
                        return response;
                    }
                }
            }
            throw new IOException("Client was shutdown before response was read");
        } catch (DisconnectException e) {
            if (client.active())
                throw e;
            else
                throw new IOException("Client was shutdown before response was read");

        }
    }
Invokes `client.send` followed by 1 or more `client.poll` invocations until a response is received or a
disconnection happens (which can happen for a number of reasons including a request timeout).

In case of a disconnection, an `IOException` is thrown.
If shutdown is initiated on the client during this method, an IOException is thrown.

This method is useful for implementing blocking behaviour on top of the non-blocking `NetworkClient`, use it with
care.
clients/src/main/java/org/apache/kafka/clients/NetworkClientUtils.java