Nikita-Shupletsov commented on code in PR #23483:
URL: https://github.com/apache/kafka/pull/23483#discussion_r4383468169
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/BufferPool.java:
##########
@@ -370,8 +353,148 @@ 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 an empty list otherwise, without reserving anything, joining
{@link #waiters} or recording a
+ * wait time. It never waits, so it ignores the thread's interrupt status
and leaves it untouched. Also
+ * returns an empty list once the pool is closed: the producer closes the
pool before it drains or aborts the
+ * last batches, and closing those batches still needs to grow their
streams. A stream growing
+ * mid-write, which can take a partial grant, uses {@link
#tryAllocateChunksUpTo} instead. Also returns an
+ * empty list while any thread is queued for memory, so non-blocking
callers never take memory ahead of a waiter.
+ *
+ * @param totalSize minimum total bytes of capacity required across the
returned chunks
+ * @return list of {@code ceil(totalSize / poolableSize())} {@code
ByteBuffer}s, each of capacity
+ * {@code poolableSize()}, or an empty list if the memory is not
available right now or the pool is closed
+ * @throws IllegalArgumentException if {@code totalSize <= 0}, or if the
request rounded up to
+ * whole chunks exceeds {@code totalMemory()}
+ */
+ public List<ByteBuffer> tryAllocateChunks(int totalSize) {
Review Comment:
I think this method also needs the `if (allocationMode !=
AllocationMode.INCREMENTAL)` check like in tryAllocateChunksUpTo
--
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]
