Hi Jun, I agree with the difference you pointed out, updated the wording in that section to make the point clear: the hasRoom (formula as you point out, new record's uncompressed size) applies equally to both strategies to decide if a new record fits in a batch, pessimistic using the record uncompressed size to avoid reject/split on large msg (no changes). For chunk-count estimation, I intentionally thought of using the compressed size of the new record being optimistic about the ratio, as it's just for allocation purposes (no impact on the actual batch size, no reject/split risk) and there's the growth mechanism for when ratio falls short. Makes sense? (details in the KIP section to double check)
Thanks! Lianet On Mon, Jun 15, 2026 at 2:31 PM Jun Rao via dev <[email protected]> wrote: > Hi, Lianet, > > Thanks for the reply. Just a minor comment. > > "projected batch size = batch header + (uncompressed bytes written so far + > new record's uncompressed upper-bound size) × estimated compression ratio × > safety factor (ratio and safety factor applied only when compression is > used)" > This is an implementation detail. But currently hasRoomFor() seems to use > the formula: batch header + uncompressed bytes written so far × estimated > compression ratio × safety factor ++ new record's uncompressed upper-bound > size > > Jun > > On Fri, Jun 12, 2026 at 12:18 PM Lianet Magrans <[email protected]> > wrote: > > > Thanks for the feedback Jun, > > > > JR5: agreed, update the names (turns out we were using the "incremental" > > word a lot already) > > > > JR6: Good point. Makes sense to simply reuse the same ratio-aware > > estimation that trunk already uses to determine if a batch "hasRoom" but > > including the new record. I updated the KIP with the details, it's pretty > > consistent with the current logic, let me know your thoughts. > > > > Thanks! > > > > > > On Thu, Jun 11, 2026 at 12:13 PM Jun Rao via dev <[email protected]> > > wrote: > > > > > Hi, Lianet, > > > > > > Thanks for the updated KIP. A couple of more minor comments. > > > > > > JR5. "buffer.memory.allocation.strategy" type String -> “static”, > > > “dynamic” > > > static vs dynamic doesn't quite capture the essence of the strategy. > How > > > about sth like full vs incremental? > > > > > > JR6. "The number of chunks needed for a record is calculated based on > the > > > record's uncompressed upper-bound size." > > > Currently, the allocated size is based on the record's estimated size > > using > > > the compression ratio. Do we plan to change that and if so, what's the > > > motivation behind it? > > > > > > Jun > > > > > > On Wed, May 27, 2026 at 12:48 PM Lianet Magrans <[email protected]> > > > wrote: > > > > > > > Thanks for the review Jun! Fixed. > > > > > > > > Cheers, > > > > Lianet > > > > > > > > On Wed, May 27, 2026 at 1:40 PM Jun Rao via dev < > [email protected]> > > > > wrote: > > > > > > > > > Hi, Lianet, > > > > > > > > > > Thanks for the updated KIP. Just a minor comment. "but the key > > > different" > > > > > should be "but the key difference". > > > > > > > > > > Jun > > > > > > > > > > On Fri, May 22, 2026 at 5:22 AM Lianet Magrans <[email protected] > > > > > > wrote: > > > > > > > > > > > Hi Jun, thanks for the feedback! (sorry for the delay, > > > > > Current/travelling) > > > > > > > > > > > > JR1: Agreed, I updated the example (aligned with the same mixed > > > > workload > > > > > > case mentioned at the beginning of the motivation) > > > > > > > > > > > > JR3: The "scale down" was referring to the reservation only > (memory > > > > held > > > > > by > > > > > > open batches, less during low-traffic vs high traffic period). > > > > Clarified > > > > > in > > > > > > the KIP to make clear that it's not about pool memory scaling > down, > > > > just > > > > > > about reservation in open batches (pool memory free for other > > > > partitions > > > > > if > > > > > > needed). > > > > > > > > > > > > JR4: Yes, it was confusing indeed. The intention was just to > refer > > to > > > > the > > > > > > producer thread marking the batch for closing (not the actual > > close). > > > > > This > > > > > > will all be the same as today when the batch fills up, as you > > > described > > > > > > (producer just "marks for close", sender does the actual close > and > > > > frees > > > > > > memory up). I clarified it all in the KIP to be accurate. > > > > > > > > > > > > Thanks! > > > > > > Lianet > > > > > > > > > > > > > > > > > > On Fri, May 15, 2026 at 9:39 PM Jun Rao via dev < > > > [email protected]> > > > > > > wrote: > > > > > > > > > > > > > Hi, Lianet, > > > > > > > > > > > > > > Thanks for the reply. > > > > > > > > > > > > > > JR1. "Memory usage: under the current static strategy, a > producer > > > > > writing > > > > > > > 10 MiB/s of aggregate throughput to a 1000-partition topic with > > > > > > > RoundRobinPartitioner struggles to achieve a meaningful > fraction > > of > > > > > that > > > > > > at > > > > > > > the default 16384 bytes "batch.size". Each partition only sends > > > 16384 > > > > > > bytes > > > > > > > at a time over a high-latency link, so per-partition throughput > > is > > > > > > bounded > > > > > > > by "16384 bytes / RTT". Increasing "batch.size" to 4 MiB > unblocks > > > > > > > throughput but the producer would need 4 MiB × 1000 partitions > = > > 4 > > > > GiB > > > > > of > > > > > > > pool memory (regardless of actual volume of data flowing per > > > > > partition). > > > > > > > Under the dynamic strategy and the same batch.size = 4 MiB, > > target > > > > > > > throughput of ~10 KiB/s and linger.ms = 100ms, per-partition > > > memory > > > > > > > becomes > > > > > > > ≈ ~1 KiB (10 KiB throughput × 100 ms linger), so total memory ≈ > > > 1000 > > > > × > > > > > 1 > > > > > > > KiB = ~1 MiB (orders of magnitude less than the 4 GiB used > under > > > the > > > > > > static > > > > > > > allocation). Similar savings apply to any workload where > > per-batch > > > > data > > > > > > > falls short of batch.size: hot-cold partition distributions > > (skewed > > > > key > > > > > > > traffic), bursty workloads with quiet periods, and > > over-provisioned > > > > > > > batch.size settings." > > > > > > > This example is still not very convincing. It's true that one > can > > > set > > > > > > > batch.size=4MB without running out of memory, but it doesn't > > > achieve > > > > > the > > > > > > > batching benefit. So, why will a user bother setting a high > batch > > > > size? > > > > > > One > > > > > > > possible example is a client that publishes to a high volume > > topic > > > > > > without > > > > > > > keys, and to a low-volume topic with keys, using the default > > > > > partitioning > > > > > > > strategy. When a high batch size is set, the static approach > may > > > > > exhaust > > > > > > > the buffer pool, whereas the dynamic approach avoids exhausting > > the > > > > > pool > > > > > > > and still achieves the batching benefit for the high volume > > topic. > > > > > > > > > > > > > > > > > > > > > JR3. "Dynamic uses aggregate_throughput × linger.ms, which > > > operators > > > > > > > control. During lower-traffic periods, static still reserves > 400 > > > MiB > > > > > > until > > > > > > > batches close; dynamic scales down proportionally." > > > > > > > Hmm, if the dynamic approach ever allocates 400MB worth of > > chunks, > > > it > > > > > > never > > > > > > > deallocates them right? Then, how will dynamic scale down? > > > > > > > > > > > > > > > > > > > > > JR4. "If the non-blocking acquire fails (pool exhausted), the > > > > producer > > > > > > will > > > > > > > close the current batch (making it eligible to drain), and > blocks > > > on > > > > > the > > > > > > > pool to allocate the chunks for the new record (up to > > max.block.ms > > > > )." > > > > > > > To be precise, currently, when the buffer pool is exhausted, > the > > > > > producer > > > > > > > doesn't close the batch directly. The background sender thread > > > drains > > > > > and > > > > > > > closes the batch. > > > > > > > > > > > > > > Jun > > > > > > > > > > > > > > On Thu, May 14, 2026 at 2:30 PM Lianet Magrans < > > [email protected] > > > > > > > > > > wrote: > > > > > > > > > > > > > > > Hi Jun, > > > > > > > > > > > > > > > > JR1: The example's point was about the case where flow > remains > > > > under > > > > > > the > > > > > > > > batch limit (those are the cases where we would get > significant > > > > > memory > > > > > > > > improvement/differences). But I do get your point and agree: > in > > > > > > scenarios > > > > > > > > where the full batch is used, the dynamic strategy would end > up > > > > using > > > > > > the > > > > > > > > same amount of memory. Still, in those cases the value comes > > from > > > > the > > > > > > > > predictability/tuning of the buffer.memory (memory > consumption > > > > > depends > > > > > > on > > > > > > > > known factors, not workload-dependant ones). I clarified the > > > first > > > > > > > example, > > > > > > > > and added a second one to showcase the case where it's not > > about > > > > > memory > > > > > > > > gains but about predictability. > > > > > > > > > > > > > > > > JR2: The main concern with keeping the same close-and-block > as > > > > trunk > > > > > in > > > > > > > > this case was the change it would bring into the send() > > blocking > > > > > > pattern. > > > > > > > > On trunk, send only blocks for memory for the first record > of a > > > > > batch, > > > > > > > but > > > > > > > > never mid-batch. Applying this close-and-block to the dynamic > > > > > strategy > > > > > > > > would change this (send() could block on any record > regardless > > of > > > > an > > > > > > open > > > > > > > > batch). I leaned initially toward avoiding changing the > > blocking > > > > > > > behaviour > > > > > > > > (and pay the extra direct allocation with visibility), but on > > > > second > > > > > > > > thoughts I agree it's cleaner to surface the situation to the > > API > > > > > > > (blocking > > > > > > > > on send, aligned with what trunk does on new batch only, and > > > > dynamic > > > > > > > would > > > > > > > > do at the record level). It's no change to the send or > > > max.bloc.ms > > > > > > > > < > > > > > > > > > > > > > > > > > > > > > > > > > > > > https://urldefense.com/v3/__http://max.bloc.ms__;!!Ayb5sqE7!tij1l2481b8WaW403I3sX_JjzjVmH3SFTImLH03m5t8v-95dzvTWsUQaSx4vXv1j93EM6pqXPd0mg4my$ > > > > > > > > > > > > > > > > contract really, just a different pattern that seems sensible > > > given > > > > > the > > > > > > > > "on-demand" allocation. I updated the KIP with this, and > left a > > > > > > rejected > > > > > > > > alternative for the record. Also, with this I opted for > > dropping > > > > the > > > > > > new > > > > > > > > metric I had (which was mainly to have visbility over this > new > > > > > > > > direct-allocation path, now removed) > > > > > > > > > > > > > > > > Hi TengYao: > > > > > > > > > > > > > > > > TYC1: interesting point, agree that your suggested metric > would > > > > give > > > > > > > > visibility on what's actually allocated from the pool (which > is > > > > > dynamic > > > > > > > > now, didn't make too much sense before because it was > "static", > > > > > > > > ~batch.size). I believe that for some of the scenarios you > > > shared, > > > > we > > > > > > > would > > > > > > > > be covered with the metrics that already exist in trunk > (e.g., > > > > > > > > bufferpool-wait-*, buffer-available/total-bytes, > > batch-size-avg), > > > > > > still, > > > > > > > > it's a fact that the new strategy allocates differently from > > the > > > > > pool, > > > > > > > > dynamically, and only a metric like you suggest would let us > > see > > > > how > > > > > > that > > > > > > > > goes (batch-size-avg is the closest but is post-compression > so > > > not > > > > > the > > > > > > > > same). I just wonder if it would make more sense to represent > > it > > > in > > > > > > > bytes, > > > > > > > > rather than in chunks?? (e.g, "batch-pool-bytes-avg"). It > would > > > > align > > > > > > > > better with existing metrics in this space, all in bytes. > Also > > I > > > > > expect > > > > > > > > operators probably think in bytes (not a new "chunk" concept, > > > which > > > > > is > > > > > > > just > > > > > > > > an internal implementation grouping bytes), and maybe better > > not > > > to > > > > > > > expose > > > > > > > > chunk as a unit of measure to make sure the metric ages well > > even > > > > if > > > > > > the > > > > > > > > internal chunk details move). What do you think? Will wait to > > > hear > > > > > back > > > > > > > and > > > > > > > > align before updating the KIP > > > > > > > > > > > > > > > > Thanks both! > > > > > > > > Cheers, > > > > > > > > Lianet > > > > > > > > > > > > > > > > > > > > > > > > On Wed, May 13, 2026 at 4:40 PM Jun Rao via dev < > > > > > [email protected]> > > > > > > > > wrote: > > > > > > > > > > > > > > > >> Hi, Lianet, > > > > > > > >> > > > > > > > >> Thanks for the reply. > > > > > > > >> > > > > > > > >> JR1. "As an example: a producer writing 10 MiB/s of > aggregate > > > > > > throughput > > > > > > > >> to > > > > > > > >> a 1000-partition topic with RoundRobinPartitioner struggles > to > > > > > > achieve a > > > > > > > >> meaningful fraction of that at the default 16384 bytes > > > > "batch.size". > > > > > > > Each > > > > > > > >> partition only sends 16384 bytes at a time over a > high-latency > > > > link, > > > > > > so > > > > > > > >> per-partition throughput is bounded by "16384 bytes / RTT". > > > > > Increasing > > > > > > > >> "batch.size" to 4 MiB unblocks throughput but the producer > > would > > > > > need > > > > > > 4 > > > > > > > >> MiB > > > > > > > >> × 1000 partitions = 4 GiB of pool memory to accommodate all > > > > > partitions > > > > > > > >> simultaneously (regardless of actual volume of data flowing > > per > > > > > > > >> partition)." > > > > > > > >> This example does not seem strong. In this case, the > producer > > > > still > > > > > > > >> requires 4GB of memory even with the proposed KIP to achieve > > > high > > > > > > > >> throughput because all 1000 partitions are active. > > > > > > > >> > > > > > > > >> JR2. "When a new record arrives mid-batch and the pool is > > > > exhausted, > > > > > > it > > > > > > > >> will perform direct heap allocation to allocate all the > chunks > > > > > > estimated > > > > > > > >> needed for the record uncompressed size." > > > > > > > >> Why do we need to introduce this new case for direct > > allocation? > > > > > This > > > > > > > case > > > > > > > >> exists in the static allocation approach. If the buffer pool > > is > > > > > > > exhausted, > > > > > > > >> the send() call blocks but all pending batches become > > drainable > > > to > > > > > > > prevent > > > > > > > >> deadlock. Is there any issue with using the same mechanism > for > > > > > dynamic > > > > > > > >> allocation? > > > > > > > >> > > > > > > > >> Jun > > > > > > > >> > > > > > > > >> > > > > > > > >> On Wed, May 13, 2026 at 8:53 AM Lianet Magrans < > > > > [email protected]> > > > > > > > >> wrote: > > > > > > > >> > > > > > > > >> > Hi Jun, > > > > > > > >> > > > > > > > > >> > JR1: Agreed, I updated the motivation section to clarify > the > > > > > > different > > > > > > > >> > scenarios based on keys and partitioner, and under which > > > > > situations > > > > > > it > > > > > > > >> > becomes problematic. > > > > > > > >> > > > > > > > > >> > JR2: The KIP preserves the 2 existing direct allocation > > > triggers > > > > > you > > > > > > > >> > mentioned (compressed data exceeding allocation and batch > > > > split), > > > > > > and > > > > > > > >> also > > > > > > > >> > introduces a new one (on new record mid-batch when pool > > > > exhausted, > > > > > > > >> > basically due to the per-record reservation approach). To > > > > > mitigate, > > > > > > > >> direct > > > > > > > >> > allocation is limited to one record's worth of growth per > > > batch > > > > > > (batch > > > > > > > >> > closed right after it), and we're also introducing the new > > > > metric > > > > > to > > > > > > > >> have > > > > > > > >> > visblity and allow to tune buffer.memory. Under normal > pool > > > > > > > conditions, > > > > > > > >> > direct allocations with the new strategy should happen > less > > > > often > > > > > > than > > > > > > > >> with > > > > > > > >> > the current behaviour, mainly because of the proposed > > > > improvement > > > > > to > > > > > > > try > > > > > > > >> > the pool first, non-blocking before falling back to heap > > > > > > allocation. I > > > > > > > >> > clarified it all in the Internal allocation strategy > section > > > > > > > (extending > > > > > > > >> on > > > > > > > >> > new sections "Blocking behaviour" and "Direct heap > > > allocation"). > > > > > > > Please > > > > > > > >> > take a look and let me know. > > > > > > > >> > > > > > > > > >> > Thanks for the review! > > > > > > > >> > Lianet > > > > > > > >> > > > > > > > > >> > PS: addressing TengYao's feedback shortly, thanks! > > > > > > > >> > > > > > > > > >> > On Tue, May 12, 2026 at 11:31 AM TengYao Chi < > > > > > [email protected] > > > > > > > > > > > > > > >> > wrote: > > > > > > > >> > > > > > > > > >> > > Hi Lianet, > > > > > > > >> > > > > > > > > > >> > > Thanks for this great KIP. > > > > > > > >> > > > > > > > > > >> > > TYC1. I have one consideration regarding observability: > Do > > > we > > > > > > need a > > > > > > > >> new > > > > > > > >> > > metric for average-chunks-per-batch? With the > introduction > > > of > > > > > the > > > > > > > >> > > chunked-buffer strategy, memory usage per partition is > no > > > > > longer a > > > > > > > >> fixed > > > > > > > >> > > batch.size. While this significantly improves memory > > > > efficiency, > > > > > > it > > > > > > > >> might > > > > > > > >> > > be beneficial for operators to understand the actual > > "chunk > > > > > > > >> utilization" > > > > > > > >> > or > > > > > > > >> > > fragmentation under different workloads. Specifically, I > > > think > > > > > > this > > > > > > > >> > metric > > > > > > > >> > > would be valuable when combined with the proposed > > > > > > > bufferpool-overflow > > > > > > > >> > > metrics: it would help operators distinguish whether > > memory > > > > > > pressure > > > > > > > >> is > > > > > > > >> > > being driven by a large number of active partitions > (many > > > > small > > > > > > > >> batches) > > > > > > > >> > or > > > > > > > >> > > by individual batches becoming unexpectedly large (many > > > chunks > > > > > per > > > > > > > >> batch, > > > > > > > >> > > perhaps due to large records or low compression ratios). > > > What > > > > do > > > > > > you > > > > > > > >> > think? > > > > > > > >> > > > > > > > > > >> > > Best, > > > > > > > >> > > TengYao Chi > > > > > > > >> > > > > > > > > > >> > > On 2026/05/11 23:03:53 Jun Rao via dev wrote: > > > > > > > >> > > > Hi, Lianet, > > > > > > > >> > > > > > > > > > > >> > > > Thanks for the KIP. > > > > > > > >> > > > > > > > > > > >> > > > JR1. It would be useful to provide a bit more > motivation > > > for > > > > > the > > > > > > > >> KIP. > > > > > > > >> > The > > > > > > > >> > > > batches allocated from the buffer pool are > proportional > > to > > > > the > > > > > > > >> number > > > > > > > >> > of > > > > > > > >> > > > active partitions. For publishing records without > keys, > > > the > > > > > > active > > > > > > > >> > > > partition is 1 by default, independent of the number > of > > > > > > partitions > > > > > > > >> in a > > > > > > > >> > > > topic. It's only when publishing records with keys > that > > > the > > > > > > active > > > > > > > >> > > > partition can be the total number of partitions in a > > > topic. > > > > > So, > > > > > > a > > > > > > > >> > > possible > > > > > > > >> > > > scenario is that a client publishes records without > keys > > > to > > > > > one > > > > > > > >> topic > > > > > > > >> > > while > > > > > > > >> > > > publishing records with keys to another. > > > > > > > >> > > > > > > > > > > >> > > > JR2. "Following records appended to the batch do not > > block > > > > or > > > > > > > throw. > > > > > > > >> > They > > > > > > > >> > > > attempt non-blocking pool allocation and fall back to > > > direct > > > > > > heap > > > > > > > if > > > > > > > >> > the > > > > > > > >> > > > pool is exhausted. > > > > > > > >> > > > Ensures not blocking on pool memory while already > > holding > > > > some > > > > > > > for a > > > > > > > >> > > batch". > > > > > > > >> > > > > > > > > > > >> > > > Currently, the producer only allocates memory > exceeding > > > the > > > > > > > >> configured > > > > > > > >> > > > buffer pool size in two cases. > > > > > > > >> > > > (1) Compressed data exceeding the estimated size > > > > > > > >> > > > (2) When a batch is too large for the broker's > > > > > max.message.bytes > > > > > > > and > > > > > > > >> > gets > > > > > > > >> > > > split, each sub-batch is allocated via > > > > > > > >> ByteBuffer.allocate(initialSize) > > > > > > > >> > > > directly. > > > > > > > >> > > > > > > > > > > >> > > > With the KIP, are we introducing new cases in addition > > to > > > > the > > > > > > > above > > > > > > > >> > two? > > > > > > > >> > > > > > > > > > > >> > > > Jun > > > > > > > >> > > > > > > > > > > >> > > > > > > > > > > >> > > > > > > > > > > >> > > > On Fri, May 1, 2026 at 6:03 AM Lianet Magrans < > > > > > > [email protected] > > > > > > > > > > > > > > > >> > > wrote: > > > > > > > >> > > > > > > > > > > >> > > > > Thanks for the feedback Jaisen! I like your proposed > > > > > "static" > > > > > > > for > > > > > > > >> the > > > > > > > >> > > > > current behaviour, it aligns nicely. All updated. > > > > > > > >> > > > > > > > > > > > >> > > > > Best! > > > > > > > >> > > > > Lianet > > > > > > > >> > > > > > > > > > > > >> > > > > On Thu, Apr 30, 2026 at 4:27 PM Jaisen Mathai via > dev > > < > > > > > > > >> > > > > [email protected]> > > > > > > > >> > > > > wrote: > > > > > > > >> > > > > > > > > > > > >> > > > > > Thanks Lianet. > > > > > > > >> > > > > > > > > > > > > >> > > > > > I like the proposal. > > > > > > > >> > > > > > > > > > > > > >> > > > > > I suggest a descriptive name such as static or > fixed > > > > > instead > > > > > > > of > > > > > > > >> > > legacy > > > > > > > >> > > > > for > > > > > > > >> > > > > > the default configuration value. I think these > will > > > age > > > > > > better > > > > > > > >> > while > > > > > > > >> > > > > still > > > > > > > >> > > > > > communicating that users should strongly consider > > > using > > > > > the > > > > > > > >> > > non-default > > > > > > > >> > > > > > value of dynamic. > > > > > > > >> > > > > > > > > > > > > >> > > > > > Jaisen > > > > > > > >> > > > > > > > > > > > > >> > > > > > On Thu, Apr 30, 2026 at 8:02 AM Lianet Magrans < > > > > > > > >> [email protected] > > > > > > > >> > > > > > > > > > >> > > > > wrote: > > > > > > > >> > > > > > > > > > > > > >> > > > > > > Hi all, > > > > > > > >> > > > > > > > > > > > > > >> > > > > > > I would like to start a discussion on KIP-1332 > > that > > > > > > > proposes a > > > > > > > >> > > dynamic > > > > > > > >> > > > > > > memory allocation strategy for the Kafka > producer, > > > to > > > > > > unlock > > > > > > > >> > > > > high-latency > > > > > > > >> > > > > > > scenarios increasingly common as Kafka moves > > toward > > > > > object > > > > > > > >> > storage. > > > > > > > >> > > > > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > >> > > > > > > > > >> > > > > > > > > > > > > > > > > > > > > > > > > > > > > https://urldefense.com/v3/__https://cwiki.apache.org/confluence/display/KAFKA/KIP-1332*3A*Dynamic*memory*allocation*for*the*Kafka*producer__;JSsrKysrKys!!Ayb5sqE7!t4yI-C5BwMxJ6dMJC7tuQhu94KuolbgKXyEnl4GChJGLYY2eS4NXk-GZYlnVPnuw3ESrGwKjyPDr5Bjp0Gk$ > > > > > > > >> > > > > > > > > > > > > > >> > > > > > > Thanks! > > > > > > > >> > > > > > > Lianet > > > > > > > >> > > > > > > > > > > > > > >> > > > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > >> > > > > > > > > > >> > > > > > > > > >> > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > >
