ezhou413 commented on code in PR #23483:
URL: https://github.com/apache/kafka/pull/23483#discussion_r4319302931
##########
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
+ * wait time. It never waits, so it ignores the thread's interrupt status
and leaves it untouched. Also
+ * returns {@code null} once the pool is closed: the producer closes the
pool before it drains or aborts the
+ * last batches, and closing those batches can still need to grow their
streams.
+ *
+ * @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 {@code null} 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) {
+ int chunkSize = poolableSize();
+ int numChunks = numChunksFor(totalSize, chunkSize);
+ long memoryRequired = (long) numChunks * chunkSize;
+ List<ByteBuffer> pooled;
+
+ lock.lock();
+ try {
+ if (this.closed)
+ return null;
+ pooled = reserveChunksIfAvailable(numChunks, chunkSize,
memoryRequired);
+ if (pooled == null)
+ return null;
+ } finally {
+ try {
+ signalNextWaiterIfMemoryAvailable();
Review Comment:
Removed it
--
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]
