KREST-2746 Reflect slow responses from Kafka back to the http client

[ehumber]

This PR adds code to reflect Kafka rate limiting back to the http client.

While the produce api response from Kafka does indicate that the client is being throttled to the Kafka Produce java client, this information is not exposed to the end user of that client (ie Kafka REST), so we have to infer the throttling another way.

Kafka throttles by delaying its response back to the client. So we can assume that if we are getting a growing backlog of requests waiting to be sent to kafka then kafka is throttling REST.

When this happens (or we have a backlog of calls to Kafka for some other reason) we are obliged to send responses back to kafka in the same order as the requests arrived with us, so all we can do is add 429 responses, once we hit the "throttle from this point "queue depth, to the end of the response queue. The requests for these are not sent to kafka, and so should reduce the traffic reaching kafka, possibly enough to bring it back under the throttle limit.

If the queue depth doesn't shrink sufficiently, and instead grows to the max limit, then the connection is chopped after the grace period in my previous (disconnect clients nicely) PR has expired.

[ehumber]

@dimitarndimitrov The failing unit test I mentioned is testWriteToChunkedOutputAfterTimeout. (although the new 429 test fails in the same way too, I've not been testing that one)

[ehumber]

@dimitarndimitrov ignore the first commit, I'd not tidied up properly. This one shows the hang behaviour from easymock where the close on the mappingIterator never returns:

logs look like

** has next delayed response
THREAD EXECUTOR TRIGGERSclass io.confluent.kafkarest.response.StreamingResponse$ComposingStreamingResponse
THREAD EXECUTOR closeAll from thread executor call through to a sub close class io.confluent.kafkarest.response.StreamingResponse$ComposingStreamingResponse
CLOSE could be from either THREAD or FINALLY ->  calls through to inputStreaming closeclass io.confluent.kafkarest.response.StreamingResponse$ComposingStreamingResponse
CLOSE could be either from THREAD or FINALLY -> calls through to inputStreaming closeclass io.confluent.kafkarest.response.StreamingResponse$InputStreamingResponse
CLOSE in json stream which calls mapping iterator close
delegate not nullclass com.fasterxml.jackson.databind.MappingIterator$$EnhancerByCGLIB$$4a178d4f

Process finished with exit code 130 (interrupted by signal 2: SIGINT)

[ehumber]

and with the debugger running it looks like this, and passes the test.

There is a difference (after I said there wasn't) :)

the debug line writing to sink.

This only writes out if I have a breakpoint in the async writing to sink method (eg line 471), otherwise I see the same behaviour with and without the debugger

THREAD EXECUTOR TRIGGERSclass io.confluent.kafkarest.response.StreamingResponse$ComposingStreamingResponse
THREAD EXECUTOR closeAll from thread executor call through to a sub close class io.confluent.kafkarest.response.StreamingResponse$ComposingStreamingResponse
CLOSE could be from either THREAD or FINALLY ->  calls through to inputStreaming closeclass io.confluent.kafkarest.response.StreamingResponse$ComposingStreamingResponse
CLOSE could be either from THREAD or FINALLY -> calls through to inputStreaming closeclass io.confluent.kafkarest.response.StreamingResponse$InputStreamingResponse
[2022-07-18 14:51:09,785] DEBUG Writing to sink (io.confluent.kafkarest.response.StreamingResponse:477)
CLOSE in json stream which calls mapping iterator close
delegate not nullclass com.fasterxml.jackson.databind.MappingIterator$$EnhancerByCGLIB$$4a178d4f
delegate now closed
THREAD EXECUTOR closeAll from thread executor calling responsequeue.close
CLOSE response queue from FINALLY or THREAD class io.confluent.kafkarest.response.StreamingResponse$AsyncResponseQueue

So that's a nice big clue I'm going to go investigate :)

[ehumber]

@dimitarndimitrov @AndrewJSchofield This PR could do with merging before mid-November if possible, so that we can deal with kafka back pressure before increasing (or removing) any byte based rate limits for produce.

It would be great if I could get some feedback by the end of October.

[ehumber]

Remove the MAX_CLOSE_RETRIES

[ehumber]

This is null, so it's not an object reference that we can overwrite/edit at this point.

Need to eg return the executor service and set the original object to that value

[ehumber]

Why is this no longer private?

[ehumber]

Shouldn't be volatile

[ehumber]

Rename these to be eg DURATION or GRACE_DURATION etc, need to go look up the values atm

[ehumber]

Can't do an expect with something that returns a void, need to do the playback of the calls one by one :'(

Add a comment explaining this, because this looks bonkers.

[ehumber]

Two c's in success.

[AndrewJSchofield]

I have reviewed the code with @ehumber and am happy to approve. I do not think this is a sensible way to handle streamed REST produce requests which overrun the KafkaProducer request limits, but this does at least prevent heap exhaustion when it happens. I think there's a significant refactor due in this code at some point.

[ehumber]

Sorry I didn't manage to get this tidied up in time :(.

I made a PR from a branch in confluentinc/kafka-rest, so if my original branch from my fork goes missing when I leave, hopefully you still have the code

#1098

For this PR I recommend you don't merge it until you need to. That will most probably be when you look at rate limiting around REST produce, and potentially remove it in REST and rely on backpressure from Kafka to keep the requests at a sensible limit.