lianetm commented on code in PR #23483:
URL: https://github.com/apache/kafka/pull/23483#discussion_r4187349875
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ChunkedRecordAccumulator.java:
##########
@@ -188,10 +181,10 @@ public RecordAppendResult append(String topic,
nowMs = time.milliseconds();
} else if (appendResult.needsNewBatch() && newBatch == null) {
// The open batch is done (e.g., full, closed) so start a
new one. Size it for
- // this first record with the same estimator the full
strategy uses
- // (RecordAccumulator.append), but reserve only enough for
the record rather than
- // a whole batch.size.
- // TODO: review when compression is supported.
+ // this first record only (not a whole batch.size), to the
uncompressed upper bound
Review Comment:
we want to use the compressed estimation based on the ratio (not the
uncompressed one).
We did consider using uncompressed estimation here but ended up rejecting it
(https://cwiki.apache.org/confluence/spaces/KAFKA/pages/421957499/KIP-1332+Dynamic+memory+allocation+for+the+Kafka+producer#KIP1332%3ADynamicmemoryallocationfortheKafkaproducer-RejectedAlternatives)
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ChunkedByteBufferOutputStream.java:
##########
@@ -158,8 +171,20 @@ private void advanceWhileCurrentChunkFull() {
*/
private void advanceToNextChunk() {
if (currentChunkIndex + 1 >= chunks.size()) {
- // TODO: KAFKA-20579. With compression support, grow here instead
of throwing.
- throw new IllegalStateException("write exceeded the stream's
remaining chunk capacity");
+ // Non-blocking, and null rather than an exception when the pool
has no memory right now or has been
+ // closed (the producer closes it before draining or aborting the
last batches)
+ List<ByteBuffer> pooled = pool.tryAllocateChunks(chunkSize);
Review Comment:
passing `chunkSize` here means we'll allocate 1 chunk at a time, right? is
that the intention?
This case I expect it's to cover the compressor flushing more than what we
have reserved. That may not be rare depending on the estimations I expect, and
one a time means acquiring lock, trying the pool etc.
So should we consider trying to allocate all the chunks we need (try to get
as much as we can from the poll first, then fall back to the heap for whatever
is left). We still need this `tryAllocateChunks` that does all or nothing when
allocating from the extension path (allocateExtensionChunks), but when called
from the write path (this), what about having an approach of
"tryAppendChunksUpTo"...the needs are different imo
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/BufferPool.java:
##########
@@ -370,8 +353,92 @@ public List<ByteBuffer> allocateChunks(int totalSize, long
maxTimeToBlockMs) thr
}
}
- // Allocate raw chunks for the reserved non-pooled portion outside the
lock. On error,
- // refund all reserved bytes (memoryRequired) and let the next waiter
try.
+ return materializeChunks(pooled, numChunks, chunkSize, memoryRequired);
+ }
+
+ /**
+ * Non-blocking variant of {@link #allocateChunks} for callers that can
neither wait nor fail, such as a
+ * chunked stream growing in the middle of a record. Returns the chunks if
the memory is available right
+ * away, and {@code null} otherwise, without reserving anything, joining
{@link #waiters} or recording a
Review Comment:
> without reserving anything
what about trying to allocate as much as we can from the pool? If we do
all-or-nothing we could end up falling back to heap allocation from more than
what's really needed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
