Ensure checkPendingRequests is run when the throttle delay is marginal

[ekanth]

We have observed 10 minute pauses on consumers on fetch and it recovers after the Kafka broker resets the connection. Here are the logs from the client:

1:12:48.629 PM	
     correlationId: 67644
     expectResponse: true
     logger: kafkajs
     message: Request Fetch(key: 1, version: 11)
     namespace: Connection
     size: 143

1:12:48.630 PM	
     correlationId: 67644
     logger: kafkajs
     message: Request enqueued
     namespace: RequestQueue

1:22:48.632 PM	
     logger: kafkajs
     message: Kafka server has closed connection
     namespace: Connection

On debugging this further and reading the code, we observed that if the throttle delay is marginal (say in this case 1ms), in push, the request will not be sent immediately. When scheduleCheckPendingRequests is called, it computes this.throttleUntil - Date.now() again. When Data.now() progresses 1ms (race condition), this.throttleUntil - Date.now() is zero and checkPendingRequests() is not scheduled at all. Now this connection goes into a limbo until the brokers kills it based on connections.max.idle.ms configuration on broker (which defaults to 10 minutes).

This fixes the root cause of the same issue called out in #660.

The code changes:

• Ensure checkPendingRequests is run even when the throttle delay is marginal. When there are pending requests, we keep scheduling checkPendingRequests() periodically. I am open to changing the default interval if there are concerns.
• Ensure maybeThrottle accounts for negative values. This is extra protective and Java client does the same.
• (not specifically related to this bug but noticed it) Move the lock acquisition code out of try block to ensure we only release (in the finally) the lock after it is taken successfully
• Add a test that exposes the problem which passes with the fix.

[Nevon]

Why is it that we need a minimum timeout here? I'm thinking that if the throttle time was marginal, say 1ms, and 1ms has elapsed by the time we check it, then this.throttledUntil - Date.now() will be either zero or negative. At that point, couldn't we just set the timeout value to Math.max(scheduleAt, 0)? That would just result in the pending check request happening on the next turn around the event loop, which is what we want in this case since the throttle period has already ended and we don't want to delay the requests further.

The only problem I see with that is that we need to make sure this.throttledUntil is cleared after we've scheduled the check, so that we don't end up always doing setTimeout(..., 0) as soon as we've been throttled once.

Does that make sense?

[ekanth]

Thanks @Nevon for the quick review!

Doing Math.max(scheduleAt, 0) has the exact problem you mentioned - checkPendingRequests() and scheduleCheckPendingRequests() keep calling each other in a loop without an exit criteria. Given pending is tracking requests both because of throttling and max inflight, I felt the simplest way to implement this would be to use a minimum timeout that kicks in when pending is not empty and stop otherwise.

[ekanth]

@Nevon - would love to hear if you have any follow-up thoughts on the approach for this fix.

[Nevon]

I'm not sure I'm sold on the idea of introducing up to a 10ms delay whenever there's been marginal throttling, but it's still an improvement over the current situation. This can always be improved if it turns out that it actually happens with some amount of frequency.

[ekanth]

@Nevon - I'd appreciate another look and help in merging this PR. This issue has been happening to us and would love to merge this and have an official release that fixes this.