...

1. queued.max.requests is deprecated/removed in favor of queued.max.bytes. In this case, the conversion of existing configurations could use queued.max.bytes = queued.max.requests * socket.request.max.bytes (which is conservative, but "safe")
2. queued.max.requests is supported as an alternative to queued.max.bytes (either-or), in which case no migration is required. A default value of 0 could be used to disable the feature (by default) and runtime code would pick a queue implementation depending on which configuration parameter is provided.
3. queued.max.requests is supported in addition queued.max.bytes (both respected at the same time). In this case a default value of queued.max.bytes = Long.MAX_VALUE would -1 would maintain backwards compatible behavior.

The current naming scheme of queued.max.requests (and the proposed queued.max.bytes) may be a bit opaque. Perhaps using requestQueue.max.requests and requestQueue.max.bytes would more clearly convey the meaning to users (indicating that these settings deal with the request queue specifically, and not some other). The current queued.max.requests configuration can be retained for a few more releases for backwards compatibility.

Test Plan

• A unit test that validates the behavior of ByteBoundedBlockingQueue already exists in the codebase.
• A unit test that validates correct behaviour of RequestChannel under capacity bounds would need to be written.
• A micro-benchmark for determining the performance of the various queue implementations - https://github.com/radai-rosenblatt/kafka-benchmarks
• Stress testing a broker (heavy producer load of varying request sizes) to verify that the memory limit is honored.
• Benchmarking producer and consumer throughput before/after the change to prove that ingress/egress performance remains acceptable.

...