Refactor writeAndFlushResponseToClient

[BewareMyPower]

Motivation

#429 introduced a request timeout to avoid some requests being blocked by the previous requests. However, it changes the behavior of writeAndFlushResponseToClient.

Before #429, writeAndFlushResponseToClient removed all completed response futures in the head of responseQueue and then sent the completed response to client.

After #429, writeAndFlushResponseToClient removed all response futures, and use CompletableFuture#get(long timeout, TimeUnit unit) to wait until all futures are completed or expired.

We shouldn't wait any CompletableFuture in a CompletableFuture's callback.

Modifications

In writeAndFlushResponseToClient, only remove completed or expired response futures. Here we combine peek and boolean remove(Object o) for thread safety.

For responses that are expired or completed exceptionally, just skip them and don't send any response to client because Kafka client has a retry mechanism. When Kafka processes an expired request, it also doesn't send any response to client.

For responses that are completed normally, send the response and release the internal Netty buffers if necessary.

[BewareMyPower]

@dockerzhang @wuzhanpeng PTAL

[wuzhanpeng]

LGTM 👍