junrao commented on code in PR #23483:
URL: https://github.com/apache/kafka/pull/23483#discussion_r4211601594
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ChunkedProducerBatch.java:
##########
@@ -74,17 +75,39 @@ int extensionBytesNeeded(long timestamp, byte[] key, byte[]
value, Header[] head
* {@code ChunkedRecordAccumulator.tryAppend}, which evaluates {@link
#extensionBytesNeeded}
* itself and attaches chunks before retrying, so repeating the check for
those would size the
* record a second time on every append for no added safety.
+ * <p>
+ * The check applies only to uncompressed batches, where the pre-size and
the demand are the same
+ * upper bound. For a compressed batch the demand scales with the topic's
compression ratio
+ * estimate, which can exceed 1.0 for incompressible data and can change
between pre-sizing and
+ * batch creation, so the pre-size is only a heuristic; any overshoot is
absorbed by the stream
+ * growing mid-write (see {@link ChunkedByteBufferOutputStream}).
+ * <p>
+ * If that growth had to fall back to the heap because the pool was
exhausted, the record is still
+ * appended but the batch is then closed for record appends, so it is sent
early rather than taking
+ * further records that would keep growing on the heap past buffer.memory.
*
* @return the record's future, or null if the batch is at its batch-size
limit
- * @throws IllegalStateException if the stream was not pre-sized to hold
the batch's first record
+ * @throws IllegalStateException if the uncompressed stream was not
pre-sized to hold the batch's first record
*/
@Override
public FutureRecordMetadata tryAppend(long timestamp, byte[] key, byte[]
value, Header[] headers, Callback callback, long now) {
- if (recordCount == 0 && extensionBytesNeeded(timestamp, key, value,
headers) != 0)
+ if (recordCount == 0 && recordsBuilder.compression().type() ==
CompressionType.NONE
+ && extensionBytesNeeded(timestamp, key, value, headers) != 0)
throw new IllegalStateException(
"Unexpected append to a chunked batch whose chunks lack
capacity for the record; " +
"the stream should have been pre-sized for the
batch's first record");
- return super.tryAppend(timestamp, key, value, headers, callback, now);
+ FutureRecordMetadata future = super.tryAppend(timestamp, key, value,
headers, callback, now);
+ if (future != null && stream().fallbackAllocations() > 0) {
+ // The stream grew on the heap mid-record: the pool is exhausted,
and the builder's fullness checks
+ // trust a compression ratio estimate that has proven too low.
Close for appends so later records
+ // go to a new batch (which blocks for pool memory) instead of
growing this one further on the heap.
+ // The caller's append result reads isFull(), now true, so the
sender is woken to send it early, and
+ // the drain's close() then feeds the observed ratio back to the
estimator sooner. The compressor flush in
+ // closeForRecordAppends may fall back once more, which is bounded
since nothing is appended after
Review Comment:
fall back => fall back to heap ?
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ChunkedProducerBatch.java:
##########
@@ -74,17 +75,39 @@ int extensionBytesNeeded(long timestamp, byte[] key, byte[]
value, Header[] head
* {@code ChunkedRecordAccumulator.tryAppend}, which evaluates {@link
#extensionBytesNeeded}
* itself and attaches chunks before retrying, so repeating the check for
those would size the
* record a second time on every append for no added safety.
+ * <p>
+ * The check applies only to uncompressed batches, where the pre-size and
the demand are the same
Review Comment:
> The capacity is verified here only for the batch's first record, which is
the one append no caller checks: {@code RecordAccumulator.appendNewBatch}
creates the batch and appends straight into it, relying on the stream having
been pre-sized.
Could we combine this part with the above comment?
##########
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
Review Comment:
Should we return an empty list instead of null when the memory is not
available? Otherwise, we have to guard against an empty list.
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ChunkedProducerBatch.java:
##########
@@ -74,17 +75,39 @@ int extensionBytesNeeded(long timestamp, byte[] key, byte[]
value, Header[] head
* {@code ChunkedRecordAccumulator.tryAppend}, which evaluates {@link
#extensionBytesNeeded}
* itself and attaches chunks before retrying, so repeating the check for
those would size the
* record a second time on every append for no added safety.
+ * <p>
+ * The check applies only to uncompressed batches, where the pre-size and
the demand are the same
+ * upper bound. For a compressed batch the demand scales with the topic's
compression ratio
+ * estimate, which can exceed 1.0 for incompressible data and can change
between pre-sizing and
+ * batch creation, so the pre-size is only a heuristic; any overshoot is
absorbed by the stream
+ * growing mid-write (see {@link ChunkedByteBufferOutputStream}).
+ * <p>
+ * If that growth had to fall back to the heap because the pool was
exhausted, the record is still
Review Comment:
> This batch never acquires memory itself: the capacity is arranged before
the append
The existing comment seems no longer true since in the compressed case, the
batch could acquire more memory.
##########
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);
+ if (pooled != null) {
+ ByteBuffer next = pooled.get(0);
Review Comment:
Should we assert that pooled has only 1 element?
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/ChunkedRecordAccumulatorTest.java:
##########
@@ -169,25 +179,566 @@ public void testMidBatchExtensionGrowsExistingBatch()
throws Exception {
}
/**
- * Pool that adds one append right after a chunk allocation returns,
mocking a concurrent
- * appender racing the same batch.
+ * End-to-end compression on the incremental path: records appended with
each codec build a valid
+ * compressed batch (the flatten-close writes the header and CRC over the
compressed buffer) that
+ * declares the codec on the wire and decodes back to exactly the bytes
appended.
+ */
+ @ParameterizedTest
+ @ValueSource(strings = {"none", "gzip", "snappy", "lz4", "zstd"})
+ public void testCompressedRecordsRoundTripThroughChunkedBatch(String
codec) throws Exception {
+ int chunkSize = 256;
+ Compression compression =
Compression.of(CompressionType.forName(codec)).build();
+ ChunkedRecordAccumulator accum = newAccumulator(8192, chunkSize, 64L *
chunkSize, compression);
+
+ // Enough sizeable records that the batch spans several chunks, so the
compressor writes
+ // across chunk boundaries rather than fitting in the first chunk.
+ int recordCount = 20;
+ List<byte[]> values = new ArrayList<>();
+ for (int i = 0; i < recordCount; i++) {
+ byte[] value = new byte[300];
+ Arrays.fill(value, (byte) i);
+ values.add(value);
+ accum.append(topic, partition1, i, key, value,
Record.EMPTY_HEADERS, null,
+ maxBlockTimeMs, time.milliseconds(), cluster);
+ }
+
+ Deque<ProducerBatch> dq = batchesFor(accum, tp1);
+ assertEquals(1, dq.size());
+ ProducerBatch batch = dq.peekFirst();
+ assertNotNull(batch);
+ assertEquals(recordCount, batch.recordCount);
+
+ // Finalize the batch: the flatten-close path writes the header + CRC
over the (compressed)
+ // contiguous buffer.
+ batch.close();
+ MemoryRecords records = batch.records();
+
+ // The built batch must declare the configured codec on the wire.
+ for (RecordBatch rb : records.batches())
+ assertEquals(compression.type(), rb.compressionType());
+
+ // Every record must decode back to exactly the bytes appended, in
order.
+ int i = 0;
+ for (Record r : records.records()) {
+ assertArrayEquals(key, readBytes(r.key()));
+ assertArrayEquals(values.get(i), readBytes(r.value()));
+ i++;
+ }
+ assertEquals(recordCount, i, "all appended records must be present");
+
+ accum.deallocate(batch);
+ accum.close();
+ }
+
+ /**
+ * A topic whose data doesn't compress drives its compression ratio
estimate above 1.0 (from the
+ * initial 1.0, a single such batch, whose observed ratio is just above
1.0 once the batch header is
+ * counted, raises it by at least COMPRESSION_RATIO_DETERIORATE_STEP). The
first record of a new batch
+ * must still append: its chunks are pre-sized to the uncompressed upper
bound, and any compressor
+ * overshoot must be absorbed by mid-write growth rather than rejected up
front. The estimate here is
+ * set straight to 1.3, above the ~1.0 this batch then observes, so
closing the batch lowers it by
+ * COMPRESSION_RATIO_IMPROVING_STEP to exactly 1.295 rather than raising
it.
+ */
+ @ParameterizedTest
+ @ValueSource(strings = {"gzip", "snappy", "lz4", "zstd"})
+ public void
testLargeFirstRecordAppendsWhenCompressionRatioEstimateAboveOne(String codec)
throws Exception {
+ int chunkSize = 256;
+ Compression compression =
Compression.of(CompressionType.forName(codec)).build();
+ ChunkedRecordAccumulator accum = newAccumulator(8192, chunkSize, 1024L
* chunkSize, compression);
+ CompressionRatioEstimator.setEstimation(topic, compression.type(),
1.3f);
+ try {
+ // Random bytes don't compress, and the record spans many chunks
so the estimate's
+ // inflation isn't absorbed by rounding up to a whole chunk.
+ byte[] value = new byte[20_000];
+ new Random(42).nextBytes(value);
+ accum.append(topic, partition1, 0L, key, value,
Record.EMPTY_HEADERS, null,
+ maxBlockTimeMs, time.milliseconds(), cluster);
+
+ Deque<ProducerBatch> dq = batchesFor(accum, tp1);
+ assertEquals(1, dq.size());
+ ProducerBatch batch = dq.peekFirst();
+ assertNotNull(batch);
+ assertEquals(1, batch.recordCount);
+
+ batch.close();
+ Record record = batch.records().records().iterator().next();
Review Comment:
Should we verify that the compressor actually overshoots?
##########
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.
Review Comment:
can still need to => still need to ?
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ChunkedRecordAccumulator.java:
##########
@@ -286,8 +279,13 @@ public RecordAppendResult append(String topic,
/**
* Mid-batch extension: the open batch can still take this record so grow
it in place. The
- * acquire is non-blocking and fails fast when the pool is exhausted,
closing
- * {@code batchToExtend} for appends so the record retries on the
new-batch path (blocks for memory)
+ * acquire ({@link BufferPool#tryAllocateChunks}) is non-blocking and
fails fast when the pool is
+ * exhausted, closing {@code batchToExtend} for appends so the record
retries on the new-batch path
+ * (blocks for memory).
+ * <p>
+ * {@code tryAllocateChunks} also refuses once the pool is closed. That is
handled the same way: when the
+ * retry then needs a new batch, its blocking acquire throws {@link
KafkaException} for the closed pool
Review Comment:
Claude found the following.
`A closed pool is reported as memory exhaustion
(ChunkedRecordAccumulator.java:178). When the pool is closed, the extension
path sets nonBlockingMemoryAllocationDenied = true and loops.
throwIfNoMoreRetriesAllowed runs before tryAppend's closed check. So if the
deadline has already passed, the send fails with BufferExhaustedException and
increments the buffer-exhausted metric, instead of KafkaException("Producer
closed …"). This also changes what the caller sees: BufferExhaustedException is
an ApiException, so the callback fires and a failed future is returned instead
of send() throwing. The window is narrow, but the javadoc at :286-287 describes
a different order than the code.`
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/ChunkedRecordAccumulatorTest.java:
##########
@@ -169,25 +179,566 @@ public void testMidBatchExtensionGrowsExistingBatch()
throws Exception {
}
/**
- * Pool that adds one append right after a chunk allocation returns,
mocking a concurrent
- * appender racing the same batch.
+ * End-to-end compression on the incremental path: records appended with
each codec build a valid
+ * compressed batch (the flatten-close writes the header and CRC over the
compressed buffer) that
+ * declares the codec on the wire and decodes back to exactly the bytes
appended.
+ */
+ @ParameterizedTest
+ @ValueSource(strings = {"none", "gzip", "snappy", "lz4", "zstd"})
+ public void testCompressedRecordsRoundTripThroughChunkedBatch(String
codec) throws Exception {
+ int chunkSize = 256;
+ Compression compression =
Compression.of(CompressionType.forName(codec)).build();
+ ChunkedRecordAccumulator accum = newAccumulator(8192, chunkSize, 64L *
chunkSize, compression);
+
+ // Enough sizeable records that the batch spans several chunks, so the
compressor writes
+ // across chunk boundaries rather than fitting in the first chunk.
+ int recordCount = 20;
+ List<byte[]> values = new ArrayList<>();
+ for (int i = 0; i < recordCount; i++) {
+ byte[] value = new byte[300];
+ Arrays.fill(value, (byte) i);
+ values.add(value);
+ accum.append(topic, partition1, i, key, value,
Record.EMPTY_HEADERS, null,
+ maxBlockTimeMs, time.milliseconds(), cluster);
+ }
+
+ Deque<ProducerBatch> dq = batchesFor(accum, tp1);
+ assertEquals(1, dq.size());
+ ProducerBatch batch = dq.peekFirst();
+ assertNotNull(batch);
+ assertEquals(recordCount, batch.recordCount);
+
+ // Finalize the batch: the flatten-close path writes the header + CRC
over the (compressed)
+ // contiguous buffer.
+ batch.close();
+ MemoryRecords records = batch.records();
+
+ // The built batch must declare the configured codec on the wire.
+ for (RecordBatch rb : records.batches())
+ assertEquals(compression.type(), rb.compressionType());
+
+ // Every record must decode back to exactly the bytes appended, in
order.
+ int i = 0;
+ for (Record r : records.records()) {
+ assertArrayEquals(key, readBytes(r.key()));
+ assertArrayEquals(values.get(i), readBytes(r.value()));
+ i++;
+ }
+ assertEquals(recordCount, i, "all appended records must be present");
+
+ accum.deallocate(batch);
+ accum.close();
Review Comment:
What's the convention of calling resetEstimation at the end of the test?
Should it be reset if the test changes the estimator? If so, this test changes
that.
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ChunkedByteBufferOutputStream.java:
##########
@@ -325,8 +361,9 @@ void throwIfInsufficientRemaining(int requiredBytes) {
ensureNotDeallocated();
// A single write can be split across several chunks, so the required
bytes needn't be
// contiguous: only the total free space matters. Advancing here would
waste the tail of the
- // current chunk, so writes advance lazily and this only validates.
- // TODO: review with KAFKA-20579, but growth support should belong in
advanceToNextChunk, not here.
+ // current chunk, so writes advance lazily and this only validates.
Automatic growth lives in
+ // advanceToNextChunk; this method only checks the currently attached
chunks (the accumulator
+ // uses it to decide when to attach extension chunks up front).
Review Comment:
The comment is inaccurate. The accumulator never calls
throwIfInsufficientRemaining(). This method is test only. It would be useful
document that in the javadoc.
##########
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");
Review Comment:
In the uncompressed case, we should never hit this path. So, keeping the
IllegalStateException in the uncompressed case is potentially useful.
SingleByteBufferOutputStream has the same issue. So, this is an existing issue
and could be addressed in the future.
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ChunkedProducerBatch.java:
##########
@@ -74,17 +75,39 @@ int extensionBytesNeeded(long timestamp, byte[] key, byte[]
value, Header[] head
* {@code ChunkedRecordAccumulator.tryAppend}, which evaluates {@link
#extensionBytesNeeded}
* itself and attaches chunks before retrying, so repeating the check for
those would size the
* record a second time on every append for no added safety.
+ * <p>
+ * The check applies only to uncompressed batches, where the pre-size and
the demand are the same
+ * upper bound. For a compressed batch the demand scales with the topic's
compression ratio
+ * estimate, which can exceed 1.0 for incompressible data and can change
between pre-sizing and
+ * batch creation, so the pre-size is only a heuristic; any overshoot is
absorbed by the stream
+ * growing mid-write (see {@link ChunkedByteBufferOutputStream}).
+ * <p>
+ * If that growth had to fall back to the heap because the pool was
exhausted, the record is still
+ * appended but the batch is then closed for record appends, so it is sent
early rather than taking
+ * further records that would keep growing on the heap past buffer.memory.
*
* @return the record's future, or null if the batch is at its batch-size
limit
- * @throws IllegalStateException if the stream was not pre-sized to hold
the batch's first record
+ * @throws IllegalStateException if the uncompressed stream was not
pre-sized to hold the batch's first record
*/
@Override
public FutureRecordMetadata tryAppend(long timestamp, byte[] key, byte[]
value, Header[] headers, Callback callback, long now) {
- if (recordCount == 0 && extensionBytesNeeded(timestamp, key, value,
headers) != 0)
+ if (recordCount == 0 && recordsBuilder.compression().type() ==
CompressionType.NONE
+ && extensionBytesNeeded(timestamp, key, value, headers) != 0)
throw new IllegalStateException(
"Unexpected append to a chunked batch whose chunks lack
capacity for the record; " +
"the stream should have been pre-sized for the
batch's first record");
- return super.tryAppend(timestamp, key, value, headers, callback, now);
+ FutureRecordMetadata future = super.tryAppend(timestamp, key, value,
headers, callback, now);
+ if (future != null && stream().fallbackAllocations() > 0) {
+ // The stream grew on the heap mid-record: the pool is exhausted,
and the builder's fullness checks
+ // trust a compression ratio estimate that has proven too low.
Close for appends so later records
+ // go to a new batch (which blocks for pool memory) instead of
growing this one further on the heap.
+ // The caller's append result reads isFull(), now true, so the
sender is woken to send it early, and
+ // the drain's close() then feeds the observed ratio back to the
estimator sooner. The compressor flush in
Review Comment:
drain's close() => drain's ProduerBatch.close() ?
##########
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:
Is this necessary? This call never increases memory and is never called from
a waiter.
##########
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);
Review Comment:
Claude found the following issue.
`tryAllocateChunks() can starve threads blocked waiting for memory
(BufferPool.java:383). reserveChunksIfAvailable only checks free memory and
ignores the waiters queue. Mid-write growth by appenders, and by the sender
when it closes compressed batches during drain, can take freed chunks ahead of
a thread blocked in allocateChunks. Under steady pressure that thread can hit
BufferExhaustedException at max.block.ms even though memory keeps being freed.
The full strategy doesn't have this problem because it grows on the heap. One
option: refuse when !waiters.isEmpty() and use the heap instead.`
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/BufferPool.java:
##########
@@ -226,8 +226,8 @@ public ByteBuffer allocate(int size, long maxTimeToBlockMs)
throws InterruptedEx
* @return the nanos actually waited, for the caller to deduct from its
blocking budget
* @throws KafkaException if the pool was closed during the wait
* @throws BufferExhaustedException if the wait timed out; the
buffer-exhausted metric is recorded
- * only when {@code recordExhaustedOnTimeout} is true (the
incremental extension path
- * recovers without dropping the record, so it records the drop
itself if needed)
+ * only when {@code recordExhaustedOnTimeout} is true (the
incremental strategy's new-batch
+ * caller in {@link ChunkedRecordAccumulator} records the drop
itself)
Review Comment:
What's "the drop"?
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/ChunkedRecordAccumulatorTest.java:
##########
@@ -23,11 +23,14 @@
import org.apache.kafka.common.Node;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.compress.Compression;
+import org.apache.kafka.common.errors.RecordBatchTooLargeException;
import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.metrics.KafkaMetric;
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.record.TimestampType;
+import org.apache.kafka.common.record.internal.CompressionRatioEstimator;
Review Comment:
In RecordAccumulatorTest, the large-append, split and ratio tests still run
only with the FULL strategy. Should we extend that to INCREMENTAL too?
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ChunkedProducerBatch.java:
##########
@@ -74,17 +75,39 @@ int extensionBytesNeeded(long timestamp, byte[] key, byte[]
value, Header[] head
* {@code ChunkedRecordAccumulator.tryAppend}, which evaluates {@link
#extensionBytesNeeded}
* itself and attaches chunks before retrying, so repeating the check for
those would size the
* record a second time on every append for no added safety.
+ * <p>
+ * The check applies only to uncompressed batches, where the pre-size and
the demand are the same
+ * upper bound. For a compressed batch the demand scales with the topic's
compression ratio
+ * estimate, which can exceed 1.0 for incompressible data and can change
between pre-sizing and
+ * batch creation, so the pre-size is only a heuristic; any overshoot is
absorbed by the stream
+ * growing mid-write (see {@link ChunkedByteBufferOutputStream}).
+ * <p>
+ * If that growth had to fall back to the heap because the pool was
exhausted, the record is still
+ * appended but the batch is then closed for record appends, so it is sent
early rather than taking
+ * further records that would keep growing on the heap past buffer.memory.
*
* @return the record's future, or null if the batch is at its batch-size
limit
- * @throws IllegalStateException if the stream was not pre-sized to hold
the batch's first record
+ * @throws IllegalStateException if the uncompressed stream was not
pre-sized to hold the batch's first record
*/
@Override
public FutureRecordMetadata tryAppend(long timestamp, byte[] key, byte[]
value, Header[] headers, Callback callback, long now) {
- if (recordCount == 0 && extensionBytesNeeded(timestamp, key, value,
headers) != 0)
+ if (recordCount == 0 && recordsBuilder.compression().type() ==
CompressionType.NONE
Review Comment:
Claude suggests that recordsBuilder.compression().type() ==
CompressionType.NONE can be replaced with isCompressed() (in ProducerBatch).
--
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]
