This is an automated email from the ASF dual-hosted git repository.
AndrewJSchofield pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 805eb7b260b KAFKA-20723: Bound decompression memory in
ShareGroupDLQRecordFetcher (#22905)
805eb7b260b is described below
commit 805eb7b260be7d4670ad39180c99aae3d991b4ab
Author: Sushant Mahajan <[email protected]>
AuthorDate: Thu Jul 23 21:16:03 2026 +0530
KAFKA-20723: Bound decompression memory in ShareGroupDLQRecordFetcher
(#22905)
### What
A compressed source record could decompress to far more data than its
on-wire size suggests, since the previous implementation fully
materialized a compressed batch via `DefaultRecordBatch.iterator()` and
allocated per-record buffers based on an unchecked, attacker-controlled
length.
### How
Compressed v2 batches are now decompressed into a size-capped
buffer before any record is parsed, so even a single oversized record
can't force a large allocation; the cap is derived from the DLQ topic's
own `max.message.bytes` rather than an arbitrary constant. Uncompressed
batches are left uncapped (no expansion risk), and fetch()/resume() now
catch Throwable so an OOM can't escape uncontained.
### Testing
Added new unit tests in `ShareGroupDLQRecordFetcherTest` and integ tests
in `ShareConsumerDLQTest`.
Co-Authored-By: Claude Sonnet 5 <[email protected]> Reviewers:
Andrew Schofield <[email protected]>
---------
Co-authored-by: Claude Sonnet 5 <[email protected]>
---
.../clients/consumer/ShareConsumerDLQTest.java | 108 ++++++++++
.../share/dlq/ShareGroupDLQRecordFetcher.java | 217 +++++++++++++++++++--
.../share/dlq/ShareGroupDLQStateManager.java | 10 +-
.../share/dlq/ShareGroupDLQRecordFetcherTest.java | 171 +++++++++++++++-
4 files changed, 489 insertions(+), 17 deletions(-)
diff --git
a/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerDLQTest.java
b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerDLQTest.java
index 2fe897fdd71..2d5f415a9e5 100644
---
a/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerDLQTest.java
+++
b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerDLQTest.java
@@ -24,6 +24,7 @@ import org.apache.kafka.clients.admin.ConfigEntry;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.admin.OffsetSpec;
import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.config.ConfigResource;
@@ -42,6 +43,7 @@ import com.yammer.metrics.core.Meter;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@@ -599,6 +601,112 @@ public class ShareConsumerDLQTest extends
ShareConsumerTestBase {
+ "chunking), count went from " + produceCountBeforeRaise + "
to " + dlqMeterCount(METRIC_DLQ_PRODUCE_TOTAL, groupId));
}
+ /**
+ * Guards against a decompression-bomb-shaped source record: a highly
compressible record that is tiny
+ * on the wire but expands to a much larger size once decompressed. The
DLQ record fetcher bounds the
+ * decompression budget it will spend copying a record by the DLQ topic's
own {@code max.message.bytes}
+ * (there is no point retaining more decompressed data than the DLQ topic
could ever accept anyway).
+ *
+ * <p>Phase 1: the DLQ topic is configured with a deliberately low {@code
max.message.bytes}. Two records
+ * with a highly compressible ~200KB payload (tiny once gzip-compressed)
are rejected with record copy
+ * enabled. The decompression budget (derived from the low limit) is
exhausted well before the payload can
+ * be fully decompressed, so the copy is skipped - but the DLQ write
itself still succeeds, headers-only
+ * (same degraded outcome as record-copy-disabled): the DLQ metrics still
fire and the records still land
+ * on the DLQ topic, just without a key/value.
+ *
+ * <p>Phase 2: the DLQ topic's {@code max.message.bytes} is raised
comfortably above the decompressed
+ * payload size, and a fresh batch of the same records is rejected. This
time the decompression budget is
+ * sufficient, so the copy succeeds and the new DLQ records carry the
original key/value.
+ */
+ @ClusterTest
+ public void
testDlqCopyRecordSkippedWhenDecompressedSizeExceedsDlqMaxMessageBytes() throws
Exception {
+ String groupId = "dlq-decompress-cap-group";
+ String sourceTopic = "dlq-decompress-cap-source";
+ String dlqTopic = "dlq.decompress-cap";
+ int recordCount = 2;
+ int payloadSize = 200_000;
+ int lowDlqMaxMessageBytes = 2_000;
+ // Comfortably above the two records' combined decompressed payload (2
* payloadSize = 400,000 bytes)
+ // plus batch/record framing overhead.
+ int highDlqMaxMessageBytes = 500_000;
+
+ try (Admin admin = createAdminClient()) {
+ admin.createTopics(Set.of(
+ new NewTopic(sourceTopic, 1, (short) 1),
+ new NewTopic(dlqTopic, 1, (short) 1)
+ .configs(Map.of(
+
TopicConfig.ERRORS_DEADLETTERQUEUE_GROUP_ENABLE_CONFIG, "true",
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG,
Integer.toString(lowDlqMaxMessageBytes)))
+ )).all().get();
+ }
+
+ alterShareAutoOffsetReset(groupId, "earliest");
+ alterShareGroupConfig(groupId,
GroupConfig.ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG, dlqTopic);
+ alterShareGroupConfig(groupId,
GroupConfig.ERRORS_DEADLETTERQUEUE_COPY_RECORD_ENABLE_CONFIG, "true");
+
+ // Highly compressible payload: tiny on the wire (comfortably under
any max.message.bytes involved),
+ // but decompresses to `payloadSize` bytes - large enough to blow past
lowDlqMaxMessageBytes once the
+ // DLQ record fetcher tries to decompress it for copying.
+ byte[] payload = new byte[payloadSize];
+ Arrays.fill(payload, (byte) 'a');
+ try (Producer<byte[], byte[]> producer =
createProducer(Map.of(ProducerConfig.COMPRESSION_TYPE_CONFIG, "gzip"))) {
+ for (int i = 0; i < recordCount; i++) {
+ producer.send(new ProducerRecord<>(sourceTopic, 0,
"key".getBytes(StandardCharsets.UTF_8), payload));
+ }
+ producer.flush();
+ }
+ rejectRecords(groupId, sourceTopic, recordCount);
+
+ // Phase 1: DLQ write succeeds (headers-only) despite the low limit -
copy is skipped, not the write.
+ waitForCondition(() -> dlqMeterCount(METRIC_DLQ_RECORD_COUNT, groupId)
== recordCount,
+ DEFAULT_MAX_WAIT_MS, 200L,
+ () -> "Expected " + recordCount + " DLQ records, was " +
dlqMeterCount(METRIC_DLQ_RECORD_COUNT, groupId));
+ List<ConsumerRecord<byte[], byte[]>> dlqRecords =
readDlqPartition(dlqTopic, 0, recordCount);
+ assertEquals(recordCount, dlqRecords.size(), "Unexpected number of
records on the DLQ topic");
+ for (ConsumerRecord<byte[], byte[]> record : dlqRecords) {
+ assertNull(record.key(), "DLQ record key should be absent - record
copy must have been skipped");
+ assertNull(record.value(), "DLQ record value should be absent -
record copy must have been skipped");
+ assertEquals(groupId, headerValue(record,
HEADER_DLQ_ERRORS_GROUP));
+ assertEquals(sourceTopic, headerValue(record,
HEADER_DLQ_ERRORS_TOPIC));
+ }
+
+ // Phase 2: point the group at a second, freshly-created DLQ topic
whose max.message.bytes is set
+ // comfortably above the decompressed payload size up front - simpler
than altering the first
+ // topic's config and waiting for it to propagate.
+ String dlqTopic2 = "dlq.decompress-cap-2";
+ try (Admin admin = createAdminClient()) {
+ admin.createTopics(Set.of(
+ new NewTopic(dlqTopic2, 1, (short) 1)
+ .configs(Map.of(
+
TopicConfig.ERRORS_DEADLETTERQUEUE_GROUP_ENABLE_CONFIG, "true",
+ TopicConfig.MAX_MESSAGE_BYTES_CONFIG,
Integer.toString(highDlqMaxMessageBytes)))
+ )).all().get();
+ }
+ alterShareGroupConfig(groupId,
GroupConfig.ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG, dlqTopic2);
+
+ try (Producer<byte[], byte[]> producer =
createProducer(Map.of(ProducerConfig.COMPRESSION_TYPE_CONFIG, "gzip"))) {
+ for (int i = 0; i < recordCount; i++) {
+ producer.send(new ProducerRecord<>(sourceTopic, 0,
"key".getBytes(StandardCharsets.UTF_8), payload));
+ }
+ producer.flush();
+ }
+ rejectRecords(groupId, sourceTopic, recordCount);
+
+ // The budget is now sufficient, so the second batch's DLQ records (on
the new topic) must carry the
+ // copied payload. The per-group metric accumulates across both DLQ
topics used by this group.
+ waitForCondition(() -> dlqMeterCount(METRIC_DLQ_RECORD_COUNT, groupId)
== recordCount * 2,
+ DEFAULT_MAX_WAIT_MS, 200L,
+ () -> "Expected " + (recordCount * 2) + " DLQ records, was " +
dlqMeterCount(METRIC_DLQ_RECORD_COUNT, groupId));
+ List<ConsumerRecord<byte[], byte[]>> secondBatchDlqRecords =
readDlqPartition(dlqTopic2, 0, recordCount);
+ assertEquals(recordCount, secondBatchDlqRecords.size(),
+ "Unexpected number of records on the second DLQ topic");
+ for (ConsumerRecord<byte[], byte[]> record : secondBatchDlqRecords) {
+ assertArrayEquals(payload, record.value(), "DLQ record value
should be the copied payload");
+ assertEquals(groupId, headerValue(record,
HEADER_DLQ_ERRORS_GROUP));
+ assertEquals(sourceTopic, headerValue(record,
HEADER_DLQ_ERRORS_TOPIC));
+ }
+ }
+
/**
* Verifies the DLQ is NOT written when it is gated off, across the gating
conditions:
* (a) no DLQ topic configured for the group;
diff --git
a/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordFetcher.java
b/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordFetcher.java
index 4d832423b02..588077ade02 100644
---
a/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordFetcher.java
+++
b/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordFetcher.java
@@ -17,13 +17,21 @@
package org.apache.kafka.server.share.dlq;
+import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.TopicIdPartition;
import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.record.TimestampType;
+import org.apache.kafka.common.record.internal.DefaultRecord;
+import org.apache.kafka.common.record.internal.DefaultRecordBatch;
+import org.apache.kafka.common.record.internal.MemoryRecords;
import org.apache.kafka.common.record.internal.Record;
import org.apache.kafka.common.record.internal.RecordBatch;
import org.apache.kafka.common.record.internal.Records;
import org.apache.kafka.common.requests.FetchRequest;
import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.internals.BufferSupplier;
+import org.apache.kafka.common.utils.internals.ByteBufferOutputStream;
+import org.apache.kafka.common.utils.internals.CloseableIterator;
import org.apache.kafka.server.share.LogReader;
import org.apache.kafka.server.storage.log.FetchIsolation;
import org.apache.kafka.server.storage.log.FetchParams;
@@ -32,6 +40,9 @@ import org.apache.kafka.storage.internals.log.LogReadResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
@@ -51,11 +62,27 @@ import java.util.concurrent.CompletableFuture;
* Offsets that cannot be read - locally or remotely - are simply absent from
the map, leaving the caller
* to produce a DLQ record with headers only for them.
*
+ * <p>{@code maxFetchBytes} only bounds the compressed, on-the-wire size of
each read; a compressed batch
+ * can still decompress into something far larger in memory. To keep a
pathologically compressible batch
+ * from ballooning broker heap usage, decompression is bounded by {@code
maxDecompressedBytes} (shared across
+ * the whole fetch): compressed batches are decompressed into a size-capped
flat buffer before any record is
+ * parsed (see {@link #decompressBounded}), so even a single record with a
fabricated huge length can't force
+ * a large allocation. A batch that would exceed the remaining budget is
simply discarded - its offsets are
+ * skipped, same as any other partial-failure case - and the fetch continues
with whatever batches follow,
+ * rather than one oversized batch poisoning the rest of the range.
+ *
* <p>Instances are single-use: create one fetcher per {@link #fetch()} call.
*/
public class ShareGroupDLQRecordFetcher {
private static final Logger log =
LoggerFactory.getLogger(ShareGroupDLQRecordFetcher.class);
+ // Size of the transfer buffer used to pull decompressed bytes out of the
codec's stream a chunk at a
+ // time (see decompressBounded). Fixed and deliberately independent of
maxDecompressedBytes: it only
+ // affects how many read() calls happen, not whether the budget check is
correct, so tying it to the
+ // (policy-configured, potentially large) budget would allocate a scratch
buffer sized by configuration
+ // rather than by the data actually being copied. Matches the precedent in
ClientTelemetryUtils.decompress.
+ private static final int DECOMPRESS_CHUNK_BYTES = 8 * 1024;
+
private final LogReader logReader;
private final Time time;
private final ShareGroupDLQRecordParameter param;
@@ -65,6 +92,9 @@ public class ShareGroupDLQRecordFetcher {
private final int recordCount;
private final long startTime;
private final Map<Long, Record> recordMap;
+ private final long maxDecompressedBytes;
+ private final BufferSupplier bufferSupplier = BufferSupplier.create();
+ private long decompressedBytes = 0;
// We are fetching data for one TopicIdPartition only. Hence, there is no
need to keep recreating
// the maxBytes map, and we can re-use a single copy. In similar vein, we
needn't clear the offsets
// map either and just update the value corresponding to the
TopicIdPartition key across iterations.
@@ -73,7 +103,8 @@ public class ShareGroupDLQRecordFetcher {
private final CompletableFuture<Map<Long, Record>> result = new
CompletableFuture<>();
private final FetchParams fetchParams;
- public ShareGroupDLQRecordFetcher(LogReader logReader, Time time,
ShareGroupDLQRecordParameter param, int maxFetchBytes) {
+ public ShareGroupDLQRecordFetcher(LogReader logReader, Time time,
ShareGroupDLQRecordParameter param,
+ int maxFetchBytes, int
maxDecompressedBytes) {
this.logReader = logReader;
this.time = time;
this.param = param;
@@ -83,6 +114,7 @@ public class ShareGroupDLQRecordFetcher {
this.startTime = time.hiResClockMs();
this.recordMap = new HashMap<>(recordCount);
this.maxBytesMap.put(tp, maxFetchBytes);
+ this.maxDecompressedBytes = maxDecompressedBytes;
this.fetchParams = new FetchParams(
FetchRequest.CONSUMER_REPLICA_ID, // -1, reading as a
consumer
-1, // replicaEpoch
@@ -102,10 +134,14 @@ public class ShareGroupDLQRecordFetcher {
public CompletableFuture<Map<Long, Record>> fetch() {
try {
runFrom(param.firstOffset());
- } catch (Exception e) {
- // Never let an unexpected error escape; skip record copy entirely.
- log.warn("Unexpected error fetching records for {}. Skipping
record copy.", param, e);
- result.complete(Map.of());
+ } catch (Throwable e) {
+ // Never let an unexpected error - including an OutOfMemoryError
from a maliciously
+ // compressible record, or an
InvalidRecordException/KafkaException from a rejected
+ // malformed one - escape. Uses complete() (not
result.complete(Map.of())) so that any
+ // records already collected from earlier batches in this call are
still returned, rather
+ // than discarding a partially-successful copy because of one bad
batch.
+ log.warn("Unexpected error fetching records for {}. Returning
records fetched so far.", param, e);
+ complete();
}
return result;
}
@@ -139,7 +175,7 @@ public class ShareGroupDLQRecordFetcher {
// completes normally; any unexpected exceptional completion is
caught by fetch().
long advanced = collect(nextOffset,
logReadResult(future.getNow(null)));
if (advanced <= nextOffset) {
- complete(); // no progress, stop
+ complete(); // no progress - stop
return;
}
nextOffset = advanced;
@@ -168,11 +204,14 @@ public class ShareGroupDLQRecordFetcher {
}
long advanced = collect(readFrom, logReadResult);
if (advanced <= readFrom) {
- complete(); // no progress, stop
+ complete(); // no progress - stop
} else {
runFrom(advanced); // resume the loop
}
- } catch (Exception e) {
+ } catch (Throwable e) {
+ // Never let an unexpected error - including an OutOfMemoryError
from a maliciously
+ // compressible record, or an
InvalidRecordException/KafkaException from a rejected
+ // malformed one - escape; return whatever was collected so far.
log.warn("Unexpected error processing records for {}. Returning
records fetched so far.", param, e);
complete();
}
@@ -197,19 +236,168 @@ public class ShareGroupDLQRecordFetcher {
/**
* Adds the records within the requested range to the map and returns the
offset to read from next
- * (never moves backwards). Records below readFrom or above endOffset are
ignored.
+ * (never moves backwards). Records below readFrom or above endOffset are
ignored. A batch that would
+ * exceed the decompression budget ({@link #maxDecompressedBytes}) is
skipped - its offsets are left
+ * unread - but the loop continues on to whatever batches follow.
*/
private long collectRecords(Records records, long readFrom) {
long nextOffset = readFrom;
for (RecordBatch batch : records.batches()) {
- for (Record record : batch) {
- // A fetch can return a batch whose base offset is below the
requested offset, so skip
- // any record at or before the read position to avoid
re-processing and dragging
- // nextOffset backwards.
+ nextOffset = collectFromBatch(batch, nextOffset);
+ }
+ return nextOffset;
+ }
+
+ private long collectFromBatch(RecordBatch batch, long readFrom) {
+ if (!batch.isCompressed()) {
+ // No decompression risk to bound: the on-wire size already bounds
what's in memory, so no
+ // cap applies (unlike the compressed cases below, capping this
would reject legitimately
+ // large uncompressed records for no safety benefit).
+ return collectUncompressed(batch, readFrom);
+ }
+ if (batch.magic() >= RecordBatch.MAGIC_VALUE_V2) {
+ return collectFromCompressedBatch(asDefaultRecordBatch(batch),
readFrom);
+ }
+ // Legacy magic v0/v1 compressed batch, effectively unseen in
practice: no bounded-buffer path
+ // available for that format, so fall back to the cumulative cap.
+ return collectWithCumulativeCap(batch, readFrom);
+ }
+
+ private long collectUncompressed(RecordBatch batch, long readFrom) {
+ long nextOffset = readFrom;
+ for (Record record : batch) {
+ // A fetch can return a batch whose base offset is below the
requested offset, so skip
+ // any record at or before the read position to avoid
re-processing and dragging
+ // nextOffset backwards.
+ if (record.offset() < readFrom) continue;
+ if (record.offset() > endOffset) return nextOffset;
+ recordMap.put(record.offset(), record);
+ nextOffset = Math.max(nextOffset, record.offset() + 1); // never
moves backwards
+ }
+ return nextOffset;
+ }
+
+ /**
+ * Materializes the batch's own on-wire bytes into a plain in-memory
{@link DefaultRecordBatch}. This is
+ * needed because a real local log read ({@code FileRecords#batches()})
returns a lazy file-channel-backed
+ * wrapper ({@code DefaultRecordBatch.DefaultFileChannelRecordBatch})
whose underlying
+ * {@code DefaultRecordBatch} is not reachable from outside {@code
org.apache.kafka.common.record.internal}
+ * (package-private constructor, protected loader) - only
already-in-memory batches (e.g. {@link
+ * MemoryRecords}) are directly a {@code DefaultRecordBatch}. The copy is
bounded by {@link
+ * RecordBatch#sizeInBytes()} - the compressed, on-wire size, itself
already bounded by this fetch's
+ * {@code maxFetchBytes} cap - so this is safe regardless of how large the
batch decompresses to.
+ */
+ private DefaultRecordBatch asDefaultRecordBatch(RecordBatch batch) {
+ if (batch instanceof DefaultRecordBatch defaultBatch) {
+ return defaultBatch;
+ }
+ ByteBuffer buffer = ByteBuffer.allocate(batch.sizeInBytes());
+ batch.writeTo(buffer);
+ buffer.flip();
+ return (DefaultRecordBatch)
MemoryRecords.readableRecords(buffer).batches().iterator().next();
+ }
+
+ /**
+ * Decompresses the batch into a size-bounded buffer first (see {@link
#decompressBounded}), then parses
+ * records out of that already-bounded buffer via the slice-based {@link
DefaultRecord#readFrom(ByteBuffer,
+ * long, long, int, Long)} - the same one the uncompressed path uses - so
a record's declared length can
+ * never cause an allocation larger than the buffer we already control.
+ */
+ private long collectFromCompressedBatch(DefaultRecordBatch batch, long
readFrom) {
+ ByteBuffer decompressed = decompressBounded(batch);
+ if (decompressed == null) {
+ // This batch alone would exceed the remaining budget: skip past
it (its offsets are left
+ // unread) rather than aborting the whole fetch, so later batches
- which may well fit - are
+ // still read.
+ return Math.max(readFrom, batch.lastOffset() + 1);
+ }
+
+ long baseOffset = batch.baseOffset();
+ long baseTimestamp = batch.baseTimestamp();
+ int baseSequence = batch.baseSequence();
+ Long logAppendTime = batch.timestampType() ==
TimestampType.LOG_APPEND_TIME ? batch.maxTimestamp() : null;
+
+ long nextOffset = readFrom;
+ while (decompressed.hasRemaining()) {
+ Record record = DefaultRecord.readFrom(decompressed, baseOffset,
baseTimestamp, baseSequence, logAppendTime);
+ // A fetch can return a batch whose base offset is below the
requested offset, so skip
+ // any record at or before the read position to avoid
re-processing and dragging
+ // nextOffset backwards.
+ if (record.offset() < readFrom) continue;
+ if (record.offset() > endOffset) return nextOffset;
+ recordMap.put(record.offset(), record);
+ nextOffset = Math.max(nextOffset, record.offset() + 1); // never
moves backwards
+ }
+ return nextOffset;
+ }
+
+ /**
+ * Decompresses the batch's record region into a flat buffer, reading in
{@link #DECOMPRESS_CHUNK_BYTES}
+ * chunks and checking the cumulative total against the remaining
decompression budget before each chunk
+ * is kept - before any record-level parsing happens. A single record with
a fabricated huge length can't
+ * cause a large allocation this way: parsing only begins once the buffer
is already fully bounded, and the
+ * buffer-based reader rejects a record whose declared length doesn't fit
what's left of it.
+ *
+ * @return the bounded, flipped buffer ready for reading, or {@code null}
if the remaining decompression
+ * budget ({@link #maxDecompressedBytes}, shared across the whole
fetch) is already exhausted.
+ */
+ private ByteBuffer decompressBounded(DefaultRecordBatch batch) {
+ long budget = maxDecompressedBytes - decompressedBytes;
+ if (budget <= 0) return null;
+
+ int chunkBytes = Math.min(batch.sizeInBytes(), DECOMPRESS_CHUNK_BYTES);
+ try (InputStream in = batch.recordInputStream(bufferSupplier);
+ ByteBufferOutputStream out = new
ByteBufferOutputStream(chunkBytes)) {
+ byte[] chunk = new byte[chunkBytes];
+ int nRead;
+ long total = 0;
+ while ((nRead = in.read(chunk, 0, chunk.length)) != -1) {
+ total += nRead;
+ if (total > budget) {
+ log.warn("Decompressed batch data for {} exceeded the {}
byte budget. " +
+ "Stopping record copy early to bound memory use.",
param, maxDecompressedBytes);
+ return null;
+ }
+ out.write(chunk, 0, nRead);
+ }
+ decompressedBytes += total;
+ out.buffer().flip();
+ return out.buffer();
+ } catch (IOException e) {
+ throw new KafkaException("Failed to decompress batch for " +
param, e);
+ }
+ }
+
+ /**
+ * Fallback used only for legacy magic v0/v1 compressed batches (no
bounded-buffer path available for
+ * that format - see {@link #asDefaultRecordBatch}): iterates records one
at a time via {@link
+ * RecordBatch#streamingIterator}, tracking a cumulative
decompressed-bytes total across the whole fetch
+ * and stopping once {@link #maxDecompressedBytes} is exceeded. Unlike
{@link
+ * #collectFromCompressedBatch}, this does not prevent a single oversized
record's initial allocation -
+ * it only bounds growth across multiple records/batches. Because that
allocation already happened (the
+ * byte count reflects real, already-materialized memory rather than data
that was rejected before use),
+ * the overshoot is not undone: {@link #decompressedBytes} stays over
budget, so later batches on this
+ * path keep getting skipped too. The rest of the current batch is skipped
- not the whole fetch - so
+ * later batches (e.g. uncompressed ones, which carry no such risk) are
still read.
+ */
+ private long collectWithCumulativeCap(RecordBatch batch, long readFrom) {
+ long nextOffset = readFrom;
+ try (CloseableIterator<Record> iterator =
batch.streamingIterator(bufferSupplier)) {
+ while (iterator.hasNext()) {
+ Record record = iterator.next();
if (record.offset() < readFrom) continue;
if (record.offset() > endOffset) return nextOffset;
+
+ decompressedBytes += record.sizeInBytes();
+ if (decompressedBytes > maxDecompressedBytes) {
+ log.warn("Decompressed record data for {} exceeded {}
bytes at offset {}. " +
+ "Skipping the rest of this batch to bound memory use.",
+ param, maxDecompressedBytes, record.offset());
+ return Math.max(nextOffset, batch.lastOffset() + 1);
+ }
+
recordMap.put(record.offset(), record);
- nextOffset = Math.max(nextOffset, record.offset() + 1); //
never moves backwards
+ nextOffset = Math.max(nextOffset, record.offset() + 1);
}
}
return nextOffset;
@@ -220,6 +408,7 @@ public class ShareGroupDLQRecordFetcher {
* that could not be read are absent from the map; the caller produces a
headers-only DLQ record for them.
*/
private void complete() {
+ bufferSupplier.close();
log.trace("Log fetch took {} ms for {} records starting at {} for {}",
time.hiResClockMs() - startTime,
recordCount, param.firstOffset(), param);
if (recordCount != recordMap.size()) {
diff --git
a/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java
b/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java
index 52753feb9d0..51e674a514f 100644
---
a/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java
+++
b/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java
@@ -794,7 +794,15 @@ public class ShareGroupDLQStateManager {
if
(!cacheHelper.isShareGroupDlqCopyRecordEnabled(param.groupId())) {
return CompletableFuture.completedFuture(Map.of());
}
- return new ShareGroupDLQRecordFetcher(logReader, time, param,
DLQ_MAX_FETCH_BYTES).fetch();
+ // Bounds decompression memory against a pathologically
compressible source record: there's
+ // no point retaining more decompressed data than the DLQ topic
could ever accept anyway, and
+ // (unlike the user's own topic config) this value isn't
controlled by whoever produced the
+ // record being copied. Falls back to DLQ_MAX_FETCH_BYTES in the
(defensive-only) case the DLQ
+ // topic isn't resolvable here - copy-record being enabled implies
one is configured in practice.
+ int maxDecompressedBytes =
cacheHelper.shareGroupDlqTopic(param.groupId())
+ .map(cacheHelper::dlqTopicMaxMessageBytes)
+ .orElse(DLQ_MAX_FETCH_BYTES);
+ return new ShareGroupDLQRecordFetcher(logReader, time, param,
DLQ_MAX_FETCH_BYTES, maxDecompressedBytes).fetch();
}
}
diff --git
a/server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordFetcherTest.java
b/server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordFetcherTest.java
index 6779e9617cc..5dac442912f 100644
---
a/server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordFetcherTest.java
+++
b/server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordFetcherTest.java
@@ -21,11 +21,16 @@ import org.apache.kafka.common.TopicIdPartition;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.compress.Compression;
import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.record.internal.DefaultRecordBatch;
import org.apache.kafka.common.record.internal.MemoryRecords;
import org.apache.kafka.common.record.internal.Record;
+import org.apache.kafka.common.record.internal.RecordBatch;
import org.apache.kafka.common.record.internal.Records;
import org.apache.kafka.common.record.internal.SimpleRecord;
import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.common.utils.internals.BufferSupplier;
+import org.apache.kafka.common.utils.internals.ByteBufferOutputStream;
+import org.apache.kafka.common.utils.internals.ByteUtils;
import org.apache.kafka.server.share.LogReader;
import org.apache.kafka.server.util.MockTime;
import org.apache.kafka.storage.internals.log.FetchDataInfo;
@@ -33,6 +38,9 @@ import org.apache.kafka.storage.internals.log.LogReadResult;
import org.junit.jupiter.api.Test;
+import java.io.ByteArrayOutputStream;
+import java.io.InputStream;
+import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.LinkedHashMap;
@@ -63,6 +71,7 @@ class ShareGroupDLQRecordFetcherTest {
private static final TopicIdPartition TOPIC_ID_PARTITION =
new TopicIdPartition(Uuid.randomUuid(), 0, "source-topic");
private static final int MAX_FETCH_BYTES = 1024 * 1024;
+ private static final int MAX_DECOMPRESSED_BYTES = 1024 * 1024;
private final LogReader logReader = mock(LogReader.class);
@@ -72,7 +81,11 @@ class ShareGroupDLQRecordFetcherTest {
}
private ShareGroupDLQRecordFetcher fetcher(ShareGroupDLQRecordParameter
param) {
- return new ShareGroupDLQRecordFetcher(logReader, MOCK_TIME, param,
MAX_FETCH_BYTES);
+ return fetcher(param, MAX_DECOMPRESSED_BYTES);
+ }
+
+ private ShareGroupDLQRecordFetcher fetcher(ShareGroupDLQRecordParameter
param, int maxDecompressedBytes) {
+ return new ShareGroupDLQRecordFetcher(logReader, MOCK_TIME, param,
MAX_FETCH_BYTES, maxDecompressedBytes);
}
private Map<Long, Record> fetch(ShareGroupDLQRecordParameter param) throws
Exception {
@@ -88,8 +101,12 @@ class ShareGroupDLQRecordFetcherTest {
// A successful read carrying the given records (offsets assigned from 0
by MemoryRecords#withRecords).
private static LogReadResult success(SimpleRecord... records) {
+ return success(Compression.NONE, records);
+ }
+
+ private static LogReadResult success(Compression compression,
SimpleRecord... records) {
return logReadResult(
- new FetchDataInfo(null,
MemoryRecords.withRecords(Compression.NONE, records)), Errors.NONE);
+ new FetchDataInfo(null, MemoryRecords.withRecords(compression,
records)), Errors.NONE);
}
// A failed read - partial-data tolerant, so it still carries a (here
empty) FetchDataInfo.
@@ -288,4 +305,154 @@ class ShareGroupDLQRecordFetcherTest {
assertTrue(result.isEmpty());
}
+
+ @Test
+ public void testCompressedBatchFetchedCorrectly() throws Exception {
+ // A normal, well within-budget compressed batch is unaffected by the
bounded decompression path.
+ whenReadAsync(done(success(Compression.gzip().build(),
+ record("k0", "v0"), record("k1", "v1"), record("k2", "v2"))));
+
+ Map<Long, Record> result = fetch(param(0L, 2L));
+
+ assertEquals(3, result.size());
+ assertRecord(result, 0L, "k0", "v0");
+ assertRecord(result, 1L, "k1", "v1");
+ assertRecord(result, 2L, "k2", "v2");
+ }
+
+ @Test
+ public void testCompressedBatchExceedingDecompressedBudgetSkipsBatch()
throws Exception {
+ // Highly compressible values so the batch is small on the wire but
decompresses well past a
+ // tiny budget - simulates a decompression-bomb-shaped batch. The
whole batch is decompressed
+ // into a bounded buffer before any record is parsed, so exceeding the
budget yields none of its
+ // records rather than a partial subset.
+ String bigValue = "x".repeat(10_000);
+ whenReadAsync(done(success(Compression.gzip().build(),
+ record("k0", bigValue), record("k1", bigValue), record("k2",
bigValue))));
+
+ Map<Long, Record> result = fetcher(param(0L, 2L), 100).fetch().get(10,
TimeUnit.SECONDS);
+
+ assertTrue(result.isEmpty(), "Expected the over-budget batch to be
skipped, yielding no records");
+ }
+
+ @Test
+ public void
testCompressedBatchExceedingDecompressedBudgetSkippedButLaterBatchStillCollected()
throws Exception {
+ // First batch is highly compressible so it alone decompresses well
past a tiny budget; the second
+ // batch is uncompressed and carries no decompression risk. Only the
first batch's offsets should be
+ // skipped - the fetch must not abort the whole range, so the second
batch is still collected.
+ String bigValue = "x".repeat(10_000);
+ MemoryRecords compressedBatch = MemoryRecords.withRecords(0L,
Compression.gzip().build(),
+ record("k0", bigValue), record("k1", bigValue), record("k2",
bigValue));
+ MemoryRecords uncompressedBatch = MemoryRecords.withRecords(3L,
Compression.NONE, record("k3", "v3"));
+
+ whenReadAsync(done(logReadResult(
+ new FetchDataInfo(null, concatBatches(compressedBatch,
uncompressedBatch)), Errors.NONE)));
+
+ Map<Long, Record> result = fetcher(param(0L, 3L), 100).fetch().get(10,
TimeUnit.SECONDS);
+
+ assertNull(result.get(0L));
+ assertNull(result.get(1L));
+ assertNull(result.get(2L));
+ assertRecord(result, 3L, "k3", "v3");
+ }
+
+ @Test
+ public void
testLegacyCompressedBatchExceedingCumulativeCapSkipsBatchButLaterBatchStillCollected()
throws Exception {
+ // Legacy magic v0/v1 batches use the cumulative-cap fallback (no
bounded-buffer path for that
+ // format). Exceeding the cap should still only skip the rest of the
offending batch, not abort the
+ // whole fetch - a later, uncompressed batch remains unaffected.
+ String bigValue = "x".repeat(10_000);
+ MemoryRecords legacyCompressedBatch =
MemoryRecords.withRecords(RecordBatch.MAGIC_VALUE_V1, 0L,
+ Compression.gzip().build(), record("k0", bigValue), record("k1",
bigValue), record("k2", bigValue));
+ MemoryRecords uncompressedBatch = MemoryRecords.withRecords(3L,
Compression.NONE, record("k3", "v3"));
+
+ whenReadAsync(done(logReadResult(
+ new FetchDataInfo(null, concatBatches(legacyCompressedBatch,
uncompressedBatch)), Errors.NONE)));
+
+ Map<Long, Record> result = fetcher(param(0L, 3L), 100).fetch().get(10,
TimeUnit.SECONDS);
+
+ assertNull(result.get(0L));
+ assertNull(result.get(1L));
+ assertNull(result.get(2L));
+ assertRecord(result, 3L, "k3", "v3");
+ }
+
+ // Concatenates the given batches' on-wire bytes into a single Records,
simulating one read returning
+ // multiple consecutive batches.
+ private static Records concatBatches(MemoryRecords... batches) {
+ int size = 0;
+ for (MemoryRecords batch : batches) {
+ size += batch.buffer().remaining();
+ }
+ ByteBuffer combined = ByteBuffer.allocate(size);
+ for (MemoryRecords batch : batches) {
+ combined.put(batch.buffer().duplicate());
+ }
+ combined.flip();
+ return MemoryRecords.readableRecords(combined);
+ }
+
+ @Test
+ public void
testSingleRecordWithFabricatedLengthRejectedWithoutLargeAllocation() throws
Exception {
+ // Distinct from
testCompressedBatchExceedingDecompressedBudgetStopsEarly (which uses a real,
+ // legitimately-compressible payload to blow a small budget): here the
batch is tiny even once
+ // decompressed - well within budget - but the single record's own
declared length lies about how
+ // much data follows it, simulating a maliciously corrupted record
rather than a genuinely large
+ // one. Verifies the record is rejected via the bounded buffer's own
remaining-capacity check
+ // (DefaultRecord.readFrom throwing InvalidRecordException) rather
than the fetcher ever attempting
+ // an allocation sized by the fabricated (here, ~1GB) declared length.
+ whenReadAsync(done(logReadResult(new FetchDataInfo(null,
corruptedLengthBatch()), Errors.NONE)));
+
+ Map<Long, Record> result = fetch(param(0L, 0L));
+
+ assertTrue(result.isEmpty(), "Corrupted record must be rejected, not
force a large allocation");
+ }
+
+ // Builds a single-record, gzip-compressed batch whose record claims a
declared body size (~1GB) far
+ // larger than what actually follows it - fabricating a corrupted length
rather than a genuinely large
+ // payload, to exercise DefaultRecord.readFrom's bounds check in isolation
from the budget check.
+ private static Records corruptedLengthBatch() throws Exception {
+ // A normal single-record gzip batch, to source real header +
compressed record bytes from.
+ MemoryRecords source =
MemoryRecords.withRecords(Compression.gzip().build(), record("k", "v"));
+ DefaultRecordBatch batch = (DefaultRecordBatch)
source.batches().iterator().next();
+
+ // Decompress the record region to plain bytes.
+ ByteArrayOutputStream plainOut = new ByteArrayOutputStream();
+ try (InputStream in =
batch.recordInputStream(BufferSupplier.NO_CACHING)) {
+ in.transferTo(plainOut);
+ }
+ byte[] plain = plainOut.toByteArray();
+
+ // Replace the record's declared body-size varint (its first byte, for
a record this small) with
+ // one that claims a huge size, far more than actually remains.
+ ByteBuffer corruptedPlain = ByteBuffer.allocate(plain.length + 8);
+ ByteUtils.writeVarint(1_000_000_000, corruptedPlain);
+ corruptedPlain.put(plain, 1, plain.length - 1);
+ corruptedPlain.flip();
+ byte[] corrupted = new byte[corruptedPlain.remaining()];
+ corruptedPlain.get(corrupted);
+
+ // Re-compress the corrupted plain bytes with the same codec.
+ ByteBufferOutputStream compressedOut = new ByteBufferOutputStream(256);
+ try (OutputStream out =
Compression.gzip().build().wrapForOutput(compressedOut,
RecordBatch.CURRENT_MAGIC_VALUE)) {
+ out.write(corrupted);
+ }
+ compressedOut.buffer().flip();
+
+ // Splice: the original header (unchanged) + newly-compressed
corrupted record region, with the
+ // batch's LENGTH field (Int32 at offset 8: size of everything after
the length field itself)
+ // patched to match the new total size.
+ ByteBuffer originalFull = ByteBuffer.allocate(batch.sizeInBytes());
+ batch.writeTo(originalFull);
+ originalFull.flip();
+ originalFull.limit(DefaultRecordBatch.RECORD_BATCH_OVERHEAD);
+
+ ByteBuffer full =
ByteBuffer.allocate(DefaultRecordBatch.RECORD_BATCH_OVERHEAD +
compressedOut.buffer().remaining());
+ full.put(originalFull);
+ full.put(compressedOut.buffer());
+ full.flip();
+ full.putInt(8, full.limit() - 12);
+
+ return MemoryRecords.readableRecords(full);
+ }
}
