...

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 codebaseexisting unit tests for ByteBoundedBlockingQueue need extension.
• A unit test that validates correct behaviour behavior 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.

...