Status

Current state: Under Discussion

Discussion thread: HERE

JIRA: KAFKA-4133

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

With KIP-74, we now have a good way to limit the size of Fetch responses, but it may still be difficult for users to control overall memory since the consumer will send fetches in parallel to all the brokers which own partitions that it is subscribed to. To give users finer control, it might make sense to add a new setting to properly limit the memory used by Fetch responses in the consumer in a similar fashion than what we already have on the producer.

Public Interfaces

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

1. buffer.memory (Long) The total bytes of memory the consumer can use to buffer fetch responses waiting to be read after being received from the server.

Proposed Changes

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


Compatibility, Deprecation, and Migration Plan

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

• 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.