ezhou413 commented on code in PR #23483:
URL: https://github.com/apache/kafka/pull/23483#discussion_r4471668150
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ChunkedProducerBatch.java:
##########
@@ -64,27 +71,50 @@ int extensionBytesNeeded(long timestamp, byte[] key, byte[]
value, Header[] head
}
/**
- * Appends the record. This batch never acquires memory itself: the
capacity is arranged before
- * the append, by the accumulator attaching chunks (see {@link
#extensionBytesNeeded} and
- * {@link #addBuffers}).
+ * Appends the record. Its capacity is pre-sized by the accumulator before
the append: the batch is
+ * created with chunks for its first record (a blocking acquire), and each
later record gets extension
+ * chunks for the shortfall reported by {@link #extensionBytesNeeded} (see
{@link #addBuffers}) through a
+ * non-blocking acquire; if the pool cannot supply them, the accumulator
closes this batch and retries the
+ * record on a new one. Pre-sizing is an exact upper bound for an
uncompressed batch. For a compressed batch
+ * it scales with the topic's compression ratio estimate, which is only a
heuristic and can change between
+ * pre-sizing and batch creation, so the backing {@link
ChunkedByteBufferOutputStream} also grows during the
+ * write, attaching all the chunks the write still needs: from the pool
without blocking, the rest from the
+ * heap. The check here therefore covers only an uncompressed batch's
first record, the one append no caller
+ * checks ({@code RecordAccumulator.appendNewBatch} appends straight into
the new batch); every later append
+ * comes through {@code ChunkedRecordAccumulator.tryAppend}, which has
already evaluated
+ * {@link #extensionBytesNeeded}.
* <p>
- * 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. Every later
append arrives through
- * {@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.
+ * If the growth fell back to the heap, the record is still appended but
the batch is then closed for
+ * record appends, so it is sent early rather than growing further 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 && !isCompressed() &&
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 ProducerBatch.close() then feeds the observed ratio
back to the estimator sooner. The
+ // compressor flush in closeForRecordAppends may fall back to the
heap once more, which is bounded since
+ // nothing is appended after it. Closing is idempotent, and
matches how the extension path closes a
+ // batch on pool exhaustion.
+ log.debug("Closing batch for {} to record appends so it is sent
early: the buffer pool was "
Review Comment:
Makes sense, thanks for the suggestion, just learned something
--
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]
