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 fetch and the memory used by CompletedFetch before 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

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.