ezhou413 commented on code in PR #23483:
URL: https://github.com/apache/kafka/pull/23483#discussion_r4126708558
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/ChunkedRecordAccumulatorTest.java:
##########
@@ -168,6 +173,133 @@ 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 (it rises by
+ * at least COMPRESSION_RATIO_DETERIORATE_STEP after a single such batch).
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.
+ */
+ @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);
Review Comment:
added a test
--
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]
