...

Update Fetcher.java to check the number of existing in-flight fetches (this is already tracked by numInFlightFetches) fetch and the memory used by CompletedFetchbefore initiating new fetch requests in createFetches(). Dividing buffer.memory by max.fetch.bytes should tell tells us how many concurrent fetches the consumer can do.


...

This KIP should be transparent to users not interested in setting this new configuration. Users wanting to take advantage of this new feature will just need to add this new settings to their consumer's properties.

Rejected Alternatives

• Count CompletedFetch as in-flight requests:


  A simpler alternative would be to consider CompletedFetch responses as in-flight requests and count their size as max.fetch.bytes bytes instead of their actually size. That way we wouldn't have to keep counting the memory used by CompletedFetch regularly. The issue is that the actually size fetched can frequently be a lot smaller than max.fetch.bytes, thus it would often overestimate the usage of buffer.memory

  Track memory usage of fetch responses:


  On the producer, we keep track of the memory used by records to be sent. The collection they are stored in (RecordAccumulator) makes this simple. In the consumer, records received are only stored in a list and the current memory usage of each item in the list is not currently tracked whereas the number of in-flight fetches is already.