KAFKA-21068: Use batch.size as the producer record builder's write limit. - #23442
chickenchickenlove wants to merge 1 commit into
Conversation
|
@lianetm Hi! |
lianetm
left a comment
lianetm
left a comment
There was a problem hiding this comment.
Thanks @chickenchickenlove ! Took a first look.
One high level thought, could we add test coverage for the batch.size=0 case? That one I expect was partly broken too but unnoticed. It should disable batch (allow the single record in but nothing else), and not wait (even if linger > 0). I expect this PR fixes the no-wait part, and that we had no test covering it
|
|
||
| final ByteBuffer batchBuffer = buffer; | ||
| RecordAppendResult appendResult = appendNewBatch(tp, dq, timestamp, key, value, headers, callbacks, | ||
| () -> MemoryRecords.builder(batchBuffer, RecordBatch.CURRENT_MAGIC_VALUE, compression, TimestampType.CREATE_TIME, 0L), |
There was a problem hiding this comment.
This fix addresses only the "full" strategy, but the "incremental" seems to have the exact same issue here
Could we fix both consistently with this PR?
There was a problem hiding this comment.
Yes, You are right!
Let me prepare another PR to address it with some additional test codes.
2 participants
The producer currently uses the allocated buffer's remaining capacity as
the
MemoryRecordsBuilderwrite limit. For an oversized record, thiscapacity can exceed
batch.size, leaving the batch unready untillinger.msexpires even though its estimated size already exceeds theconfigured batch size.
This PR passes
batch.sizeas the write limit while retaining a bufferlarge enough to hold the record.
Tests cover immediate readiness for oversized batches, linger behavior
when the compressed size estimate is below
batch.size, and preventingadditional records from being appended to oversized batches. The
existing partitioner test is also updated to reflect earlier selection
of the next partition.
Reviewers: Lianet Magrans lmagrans@confluent.io