Nikita-Shupletsov commented on code in PR #23483:
URL: https://github.com/apache/kafka/pull/23483#discussion_r4158179340
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/BufferPool.java:
##########
@@ -296,17 +296,16 @@ public List<ByteBuffer> allocateChunks(int totalSize,
long maxTimeToBlockMs) thr
throw new KafkaException("Producer closed while allocating
memory");
}
try {
- long freeListBytes = (long) free.size() * chunkSize;
- if (this.nonPooledAvailableMemory + freeListBytes >=
memoryRequired) {
- // Enough memory available to allocate the chunks needed
- while (pooled.size() < numChunks && !free.isEmpty())
- pooled.add(free.pollFirst());
- long remainingBytes = memoryRequired - (long) pooled.size() *
chunkSize;
- if (remainingBytes > 0) {
- // remainingBytes > 0 means the free list was fully
drained into `pooled`, so the
- // remainder comes entirely from non-pooled memory
(sufficient per the check above).
- this.nonPooledAvailableMemory -= remainingBytes;
- }
+ if (reserveChunksIfAvailable(pooled, numChunks, chunkSize,
memoryRequired)) {
Review Comment:
I think it would be cleaner if we returned pooled instead of passing it in
as a parameter and expecting side effects
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/BufferPool.java:
##########
@@ -296,17 +296,16 @@ public List<ByteBuffer> allocateChunks(int totalSize,
long maxTimeToBlockMs) thr
throw new KafkaException("Producer closed while allocating
memory");
}
try {
- long freeListBytes = (long) free.size() * chunkSize;
- if (this.nonPooledAvailableMemory + freeListBytes >=
memoryRequired) {
- // Enough memory available to allocate the chunks needed
- while (pooled.size() < numChunks && !free.isEmpty())
- pooled.add(free.pollFirst());
- long remainingBytes = memoryRequired - (long) pooled.size() *
chunkSize;
- if (remainingBytes > 0) {
- // remainingBytes > 0 means the free list was fully
drained into `pooled`, so the
- // remainder comes entirely from non-pooled memory
(sufficient per the check above).
- this.nonPooledAvailableMemory -= remainingBytes;
- }
+ if (reserveChunksIfAvailable(pooled, numChunks, chunkSize,
memoryRequired)) {
+ // Enough memory was available; the chunks are reserved.
+ } else if (maxTimeToBlockMs <= 0) {
Review Comment:
I think we don't need this branch, because we added tryAllocateChunks
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/ChunkedRecordAccumulatorTest.java:
##########
@@ -168,6 +178,538 @@ public void testMidBatchExtensionGrowsExistingBatch()
throws Exception {
accum.close();
}
+ /**
+ * 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 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();
+ assertArrayEquals(value, readBytes(record.value()));
+
+ // The observed ratio (about 1.0) is below the 1.3 estimate, so
the estimator improves the
+ // estimate by COMPRESSION_RATIO_IMPROVING_STEP, but never below
the observed ratio.
+ float observed = (float) batch.compressionRatio();
+ assertTrue(observed < 1.3f, "random data should not inflate by
30%, but the observed ratio was " + observed);
+ float updated = CompressionRatioEstimator.estimation(topic,
compression.type());
+ assertEquals(Math.max(1.3f -
CompressionRatioEstimator.COMPRESSION_RATIO_IMPROVING_STEP, observed), updated);
Review Comment:
because we are deplicating the logic from the CompressionRatioEstimator, it
will always be correct. maybe it makes sense to use the exact numbers here, as
we know exactly what data we send
--
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]
