...

The following option will be added to for consumers to configure (in ConsumerConfigs.java):

1. max.in.flight.fetches (Integer) Specifies the maximum number of in-flight fetches the consumer can make. Defaults to Integer.MAX_VALUE.

Proposed Changes

Update Fetcher.java to take into account the number of existing in-flight fetches before initiating a new fetch request. NetworkClient already tracks the The total number of in-flight requests is already tracked, so we'd just need to check this value when creating new fetches in Fetcher.createFetchRequests().

...