ezhou413 commented on code in PR #23483:
URL: https://github.com/apache/kafka/pull/23483#discussion_r4319432913
##########
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:
Updated
--
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]
