Nikita-Shupletsov commented on code in PR #23483:
URL: https://github.com/apache/kafka/pull/23483#discussion_r4096373840
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ChunkedByteBufferOutputStream.java:
##########
@@ -237,15 +261,32 @@ private void releaseUnusedChunks() {
if (currentChunk == null) // already deallocated; nothing attached
return;
List<ByteBuffer> unused = chunks.subList(currentChunkIndex + 1,
chunks.size());
- if (pool != null) {
- for (ByteBuffer chunk : unused)
+ for (ByteBuffer chunk : unused) {
+ boolean poolOwned = removeByIdentity(poolAllocatedChunks, chunk);
+ if (poolOwned && pool != null)
Review Comment:
is there a situation when the pool is null?
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ChunkedByteBufferOutputStream.java:
##########
@@ -237,15 +261,32 @@ private void releaseUnusedChunks() {
if (currentChunk == null) // already deallocated; nothing attached
return;
List<ByteBuffer> unused = chunks.subList(currentChunkIndex + 1,
chunks.size());
- if (pool != null) {
- for (ByteBuffer chunk : unused)
+ for (ByteBuffer chunk : unused) {
+ boolean poolOwned = removeByIdentity(poolAllocatedChunks, chunk);
+ if (poolOwned && pool != null)
pool.deallocate(chunk);
}
// Remove the released chunks from `chunks`, so they are
// not deallocated again on batch completion.
unused.clear();
}
+ /**
+ * Removes the first element identical ({@code ==}) to {@code target} from
{@code list}, returning
+ * whether it was present. Uses reference identity rather than {@link
Object#equals} because
+ * {@link ByteBuffer#equals} compares contents, which would match the
wrong chunk (e.g. two empty
+ * chunks compare equal).
+ */
+ private static boolean removeByIdentity(List<ByteBuffer> list, ByteBuffer
target) {
Review Comment:
an alternative could be using Collections.newSetFromMap(new
IdentityHashMap<>()); for poolAllocatedChunks. then we would not need this
method.
but completely up to you
##########
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:
if would be also good to test the same situation(high entropy messages with
compression rate above 1), but with the initial estimation significantly lower
than 1. so we can verify that we correctly handle the situation when we allowed
way more data into one batch than its size
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ChunkedByteBufferOutputStream.java:
##########
@@ -158,8 +161,28 @@ private void advanceWhileCurrentChunkFull() {
*/
private void advanceToNextChunk() {
if (currentChunkIndex + 1 >= chunks.size()) {
- // TODO: KAFKA-20579. With compression support, grow here instead
of throwing.
- throw new IllegalStateException("write exceeded the stream's
remaining chunk capacity");
+ ByteBuffer next = null;
+ try {
+ List<ByteBuffer> chunk = pool.allocateChunks(chunkSize, 0);
Review Comment:
I took a deeper look into allocateChunks.
It creates conditions, updates the waiters list, waits for exactly 0ms, and
so on. so there is quite a bit of unnecessary work when we pass 0 here. so,
basically, the whole else block is unnecessary:
https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/producer/internals/BufferPool.java#L310-L365
we could simplify the path there if 0 is passed. or, as claude suggested, we
could create a separate method: tryAllocateChunks, which would return null
instead of throwing BufferExhaustedException. it may be a cleaner way, but
needs a deeper look into it
--
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]
