You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org> on 2026/10/01 17:25:18 UTC
Re: [PR] KAFKA-20579: Implement compression support for KIP-1332 [kafka]
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org