...

Proposed Changes

Update Fetcher.java so before it initiate new Fetches, it checks:

• The CompletedFetch list, counting the raw size of each record
• The in-flight request, counting them as max.fetch.bytes

By adding the 2 together, we can estimate the currently used memory (usedBytes). To determine how many new fetches can be made, we can then divide the available space (buffer.memory - usedBytes) to check the number of existing in-flight fetch and the memory used by CompletedFetchbefore initiating new fetch requests in createFetches(). Dividing buffer.memory by max.fetch.bytes tells us how many concurrent fetches the consumer can do.

Compatibility, Deprecation, and Migration Plan

...