...

• Adding a new config input.buffer.max.bytes applicable at a topology level. The importance of this config would be Medium. 
  • Importance: Medium
  • Default Value: 512 MB.
• Adding new config statestore.Renaming cache.max.bytes. buffering to statestore.Importance and default value would be the same as  cache.max.bytes.buffering.

Compatibility, Deprecation, and Migration Plan

• The per partition config buffered.records.per.partition would be deprecated.
• The config  cache.max.bytes.buffering would be deprecated.

Rejected Alternatives

• Having the config input.buffer.max.bytes at a global level and not distributed across Stream Threads. In this case, all the bytes aggregated across all tasks is calculated at an instance level and when it exceeds the global config, the partitions with non-empty buffers would be paused. 
  • Pros
    1. Simplest in terms of implementation.
  • Cons
    1. The issue with this scheme could be achieving fairness across Stream Threads. Consider the following scenario: Let's say there are more than 1 Stream Threads in an instance, and only one of them is exceeding the bounds individually. Since the algorithm is looking at the overall bytes count across all tasks within the instance, the other Threads would also end up paying the penalty and be paused. Note that, fairness across Stream Tasks within a Stream Thread is not a problem in any scenario as the consumer does round robin fetching.
• Fine grained memory allocation across Tasks/Partitions. The current proposed config does memory allocation only at a Stream Thread level and stops. One of the alternatives was to assign memory not just at a Thread level but trickling it down further at a Task level and then finally at a Partition level.
  • Pros
    • Assigning memory down to the Partition level would make the behaviour closest to the current buffered.records.per.partition. 
    • We would pause only the exceeding partition instead of the proposed scheme where all non empty partitions are paused, which might lead to unwanted pauses.
  • Cons
    • The only reason this option was rejected was to keep things simple.
• Heuristic Based Pauses: Instead of pausing all non empty partitions, use some heuristics to pause the partitions like pause partitions accounting for  X % of bytes or pick the one with most bytes and pause only those. This option was also rejected in favour of simplicity.

...