You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2026/09/13 11:52:31 UTC
(kafka) branch 4.4 updated: Bound decompressed record size (#23447)
This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch 4.4
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/4.4 by this push:
new b69c07c8169 Bound decompressed record size (#23447)
b69c07c8169 is described below
commit b69c07c8169358e90bd442ecd0cfe5e72d8c4ce7
Author: Manikumar Reddy <ma...@gmail.com>
AuthorDate: Sun Sep 13 17:22:24 2026 +0530
Bound decompressed record size (#23447)
The fix improves validation and handling of record sizes across the
broker's record processing paths.
A new configurable broker/topic-level setting,
max.decompressed.message.bytes, allows operators to control the maximum
permitted decompressed record size. The limit is applied consistently
during produce validation, log compaction and timestamp-based list
offsets lookups.
Existing APIs remain backward compatible with safe default behavior for
callers that do not explicitly configure a limit.
Reviewers: Chia-Ping Tsai <ch...@gmail.com>
(cherry picked from commit 4d781f506a777c43eb93b359ed8a171e032d2df4)
Co-authored-by: Srinivas Akhil Mallela <sr...@gmail.com>
---
checkstyle/suppressions.xml | 4 +-
.../apache/kafka/common/config/TopicConfig.java | 12 ++
.../record/internal/AbstractLegacyRecordBatch.java | 36 ++++-
.../common/record/internal/DefaultRecord.java | 34 ++++-
.../common/record/internal/DefaultRecordBatch.java | 22 ++-
.../common/record/internal/FileLogInputStream.java | 5 +
.../kafka/common/record/internal/FileRecords.java | 22 ++-
.../common/record/internal/MemoryRecords.java | 28 ++--
.../common/record/internal/MutableRecordBatch.java | 9 ++
.../kafka/common/record/internal/RecordBatch.java | 16 ++-
.../kafka/common/record/internal/Records.java | 7 +
.../internals/FetchRequestManagerTest.java | 2 +-
.../clients/consumer/internals/FetcherTest.java | 2 +-
.../internals/ShareConsumeRequestManagerTest.java | 3 +-
.../internal/AbstractLegacyRecordBatchTest.java | 65 +++++++++
.../record/internal/DefaultRecordBatchTest.java | 95 +++++++++++++
.../common/record/internal/DefaultRecordTest.java | 115 +++++++++++++++-
.../common/record/internal/FileRecordsTest.java | 28 +++-
.../common/record/internal/MemoryRecordsTest.java | 22 +--
.../src/main/scala/kafka/server/ConfigHelper.scala | 3 +-
core/src/main/scala/kafka/server/KafkaConfig.scala | 3 +
.../test/scala/unit/kafka/log/LogConfigTest.scala | 21 +++
.../scala/unit/kafka/server/KafkaApisTest.scala | 34 ++++-
.../scala/unit/kafka/server/KafkaConfigTest.scala | 2 +
.../apache/kafka/server/config/ServerConfigs.java | 6 +
.../kafka/server/config/ServerLogConfigs.java | 1 +
.../server/config/ServerTopicConfigSynonyms.java | 1 +
.../org/apache/kafka/server/ConfigHelperUtils.java | 23 +++-
.../kafka/server/requests/ProduceRequestTest.java | 153 +++++++++++++++++++++
.../log/remote/storage/RemoteLogManager.java | 11 +-
.../kafka/storage/internals/log/Cleaner.java | 20 ++-
.../kafka/storage/internals/log/LogConfig.java | 14 ++
.../kafka/storage/internals/log/LogSegment.java | 6 +-
.../kafka/storage/internals/log/LogValidator.java | 25 +++-
.../kafka/storage/internals/log/UnifiedLog.java | 7 +-
.../log/remote/storage/RemoteLogManagerTest.java | 41 +++++-
.../storage/internals/log/LogCleanerTest.java | 28 ++++
.../storage/internals/log/LogSegmentTest.java | 18 +--
.../kafka/storage/internals/log/LogTestUtils.java | 5 +
.../storage/internals/log/LogValidatorTest.java | 45 ++++++
.../storage/internals/log/UnifiedLogTest.java | 86 +++++++++++-
41 files changed, 986 insertions(+), 94 deletions(-)
diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml
index b3fd9b27ade..5fca9456e41 100644
--- a/checkstyle/suppressions.xml
+++ b/checkstyle/suppressions.xml
@@ -308,9 +308,11 @@
files="(LocalLog|LogLoader|LogManager|LogValidator|RemoteLogManager|RemoteIndexCache|UnifiedLog).java"/>
<suppress checks="ParameterNumber"
files="(LogLoader|LogManager|UnifiedLog).java"/>
+ <suppress checks="ParameterNumber"
+ files="[/\\]Cleaner\.java"/>
<suppress checks="(ClassDataAbstractionCoupling|ClassFanOutComplexity)"
files="(UnifiedLog|RemoteLogManager|RemoteLogManagerTest|UnifiedLogTest).java"/>
- <suppress checks="MethodLength" files="RemoteLogManagerConfig.java"/>
+ <suppress checks="MethodLength" files="(RemoteLogManagerConfig|UnifiedLog).java"/>
<suppress checks="JavaNCSS" files="(RemoteLogManagerTest|UnifiedLogTest).java"/>
<!-- storage test -->
diff --git a/clients/src/main/java/org/apache/kafka/common/config/TopicConfig.java b/clients/src/main/java/org/apache/kafka/common/config/TopicConfig.java
index 6589ac7fac4..d79194e9a89 100755
--- a/clients/src/main/java/org/apache/kafka/common/config/TopicConfig.java
+++ b/clients/src/main/java/org/apache/kafka/common/config/TopicConfig.java
@@ -136,6 +136,18 @@ public class TopicConfig {
public static final String MAX_MESSAGE_BYTES_DOC =
"The largest record batch size allowed by Kafka (after compression if compression is enabled).";
+ public static final String MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG = "max.decompressed.message.bytes";
+ public static final String MAX_DECOMPRESSED_MESSAGE_BYTES_DOC = "The maximum decompressed size, in bytes, of a " +
+ "single record the broker will accept; larger records are rejected with an " +
+ "<code>InvalidRecordException</code>. It is enforced where the broker decompresses records: produce " +
+ "validation of compressed batches, log compaction, and offset lookups by timestamp or max timestamp against " +
+ "the local log or remote storage. " +
+ "Unlike <code>max.message.bytes</code>, which bounds the compressed batch on the wire, this bounds the " +
+ "decompressed size of an individual record. Lowering it below the size of already-stored records makes " +
+ "compaction and timestamp offset lookups of affected partitions fail until it is raised again. The " +
+ "default imposes no limit beyond the largest record the broker could allocate anyway; set a smaller value " +
+ "to enforce a per-record cap.";
+
public static final String INDEX_INTERVAL_BYTES_CONFIG = "index.interval.bytes";
public static final String INDEX_INTERVAL_BYTES_DOC = "This setting controls how frequently Kafka " +
"adds entries to its offset index and, conditionally, to its time index. " +
diff --git a/clients/src/main/java/org/apache/kafka/common/record/internal/AbstractLegacyRecordBatch.java b/clients/src/main/java/org/apache/kafka/common/record/internal/AbstractLegacyRecordBatch.java
index 2083bf16dd9..824f4000b6f 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/internal/AbstractLegacyRecordBatch.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/internal/AbstractLegacyRecordBatch.java
@@ -233,8 +233,12 @@ public abstract class AbstractLegacyRecordBatch extends AbstractRecordBatch impl
}
CloseableIterator<Record> iterator(BufferSupplier bufferSupplier) {
+ return iterator(bufferSupplier, Records.SOFT_MAX_ARRAY_LENGTH);
+ }
+
+ CloseableIterator<Record> iterator(BufferSupplier bufferSupplier, int maxRecordBodySize) {
if (isCompressed())
- return new DeepRecordsIterator(this, false, Integer.MAX_VALUE, bufferSupplier);
+ return new DeepRecordsIterator(this, false, Integer.MAX_VALUE, bufferSupplier, maxRecordBodySize);
return new CloseableIterator<>() {
private boolean hasNext = true;
@@ -268,6 +272,12 @@ public abstract class AbstractLegacyRecordBatch extends AbstractRecordBatch impl
return iterator(bufferSupplier);
}
+ @Override
+ public CloseableIterator<Record> streamingIterator(BufferSupplier bufferSupplier, int maxRecordBodySize) {
+ // the older message format versions do not support streaming, so we return the normal iterator
+ return iterator(bufferSupplier, maxRecordBodySize);
+ }
+
static void writeHeader(ByteBuffer buffer, long offset, int size) {
buffer.putLong(offset);
buffer.putInt(size);
@@ -281,11 +291,13 @@ public abstract class AbstractLegacyRecordBatch extends AbstractRecordBatch impl
private static final class DataLogInputStream implements LogInputStream<AbstractLegacyRecordBatch> {
private final InputStream stream;
private final int maxMessageSize;
+ private final int maxRecordBodySize;
private final ByteBuffer offsetAndSizeBuffer;
- DataLogInputStream(InputStream stream, int maxMessageSize) {
+ DataLogInputStream(InputStream stream, int maxMessageSize, int maxRecordBodySize) {
this.stream = stream;
this.maxMessageSize = maxMessageSize;
+ this.maxRecordBodySize = maxRecordBodySize;
this.offsetAndSizeBuffer = ByteBuffer.allocate(Records.LOG_OVERHEAD);
}
@@ -301,6 +313,15 @@ public abstract class AbstractLegacyRecordBatch extends AbstractRecordBatch impl
throw new CorruptRecordException(String.format("Record size is less than the minimum record overhead (%d)", LegacyRecord.RECORD_OVERHEAD_V0));
if (size > maxMessageSize)
throw new CorruptRecordException(String.format("Record size exceeds the largest allowable message size (%d).", maxMessageSize));
+ // The maxMessageSize check above does not bound the allocation on the compressed deep-decode
+ // path, where maxMessageSize can be up to Integer.MAX_VALUE. Reject, before allocating, any
+ // inner record whose declared size exceeds the configured per-record maximum. maxRecordBodySize
+ // never exceeds SOFT_MAX_ARRAY_LENGTH (the default here, and the config's upper bound), so this
+ // also stops an adversarial inner size from triggering an OutOfMemoryError (see the equivalent
+ // guard in DefaultRecord for the V2 format).
+ if (size > maxRecordBodySize)
+ throw new InvalidRecordException("Invalid record size: " + size +
+ " exceeds the configured maximum record size of " + maxRecordBodySize + ".");
ByteBuffer batchBuffer = ByteBuffer.allocate(size);
Utils.readFully(stream, batchBuffer);
@@ -320,7 +341,8 @@ public abstract class AbstractLegacyRecordBatch extends AbstractRecordBatch impl
private DeepRecordsIterator(AbstractLegacyRecordBatch wrapperEntry,
boolean ensureMatchingMagic,
int maxMessageSize,
- BufferSupplier bufferSupplier) {
+ BufferSupplier bufferSupplier,
+ int maxRecordBodySize) {
LegacyRecord wrapperRecord = wrapperEntry.outerRecord();
this.wrapperMagic = wrapperRecord.magic();
if (wrapperMagic != RecordBatch.MAGIC_VALUE_V0 && wrapperMagic != RecordBatch.MAGIC_VALUE_V1)
@@ -335,7 +357,7 @@ public abstract class AbstractLegacyRecordBatch extends AbstractRecordBatch impl
wrapperMagic + ")");
InputStream stream = Compression.of(compressionType).build().wrapForInput(wrapperValue, wrapperRecord.magic(), bufferSupplier);
- LogInputStream<AbstractLegacyRecordBatch> logStream = new DataLogInputStream(stream, maxMessageSize);
+ LogInputStream<AbstractLegacyRecordBatch> logStream = new DataLogInputStream(stream, maxMessageSize, maxRecordBodySize);
long lastOffsetFromWrapper = wrapperEntry.lastOffset();
long timestampFromWrapper = wrapperRecord.timestamp();
@@ -518,6 +540,12 @@ public abstract class AbstractLegacyRecordBatch extends AbstractRecordBatch impl
return CloseableIterator.wrap(iterator(bufferSupplier));
}
+ @Override
+ public CloseableIterator<Record> skipKeyValueIterator(BufferSupplier bufferSupplier, int maxRecordBodySize) {
+ // legacy batches cannot cheaply skip the record body, so this is a full decode
+ return CloseableIterator.wrap(iterator(bufferSupplier, maxRecordBodySize));
+ }
+
@Override
public void writeTo(ByteBufferOutputStream outputStream) {
outputStream.write(buffer.duplicate());
diff --git a/clients/src/main/java/org/apache/kafka/common/record/internal/DefaultRecord.java b/clients/src/main/java/org/apache/kafka/common/record/internal/DefaultRecord.java
index e8241eb0b70..3b7ff692123 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/internal/DefaultRecord.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/internal/DefaultRecord.java
@@ -277,12 +277,28 @@ public class DefaultRecord implements Record {
return result;
}
+ /**
+ * Decode a record from the (decompressed) stream, rejecting any record whose declared body size
+ * exceeds {@code maxRecordBodySize} before allocating the body buffer. Callers that do not
+ * enforce a configured limit pass {@link Records#SOFT_MAX_ARRAY_LENGTH}, effectively the
+ * array-length allocation limit.
+ */
public static DefaultRecord readFrom(InputStream input,
long baseOffset,
long baseTimestamp,
int baseSequence,
- Long logAppendTime) throws IOException {
+ Long logAppendTime,
+ int maxRecordBodySize) throws IOException {
int sizeOfBodyInBytes = ByteUtils.readVarint(input);
+ if (sizeOfBodyInBytes < 0)
+ throw new InvalidRecordException("Invalid record size: " + sizeOfBodyInBytes + " is negative.");
+ // Reject, before allocating, any record whose declared (decompressed) body exceeds the
+ // configured per-record maximum. maxRecordBodySize never exceeds SOFT_MAX_ARRAY_LENGTH (the
+ // default here, and the config's upper bound), so this also stops an adversarial size from
+ // reaching ByteBuffer.allocate and raising OutOfMemoryError.
+ if (sizeOfBodyInBytes > maxRecordBodySize)
+ throw new InvalidRecordException("Invalid record size: " + sizeOfBodyInBytes +
+ " exceeds the configured maximum record size of " + maxRecordBodySize + ".");
ByteBuffer recordBuffer = ByteBuffer.allocate(sizeOfBodyInBytes);
int bytesRead = Utils.readFully(input, recordBuffer);
if (bytesRead != sizeOfBodyInBytes)
@@ -359,12 +375,26 @@ public class DefaultRecord implements Record {
}
}
+ /**
+ * Skip-decode a record from the (decompressed) stream, rejecting any record whose declared body
+ * size exceeds {@code maxRecordBodySize}. See
+ * {@link #readFrom(InputStream, long, long, int, Long, int)}.
+ */
public static PartialDefaultRecord readPartiallyFrom(InputStream input,
long baseOffset,
long baseTimestamp,
int baseSequence,
- Long logAppendTime) throws IOException {
+ Long logAppendTime,
+ int maxRecordBodySize) throws IOException {
int sizeOfBodyInBytes = ByteUtils.readVarint(input);
+ if (sizeOfBodyInBytes < 0)
+ throw new InvalidRecordException("Invalid record size: " + sizeOfBodyInBytes + " is negative.");
+ // Reject records whose declared (decompressed) body exceeds the configured per-record
+ // maximum; as in readFrom, this doubles as the array-length allocation guard because
+ // maxRecordBodySize never exceeds SOFT_MAX_ARRAY_LENGTH.
+ if (sizeOfBodyInBytes > maxRecordBodySize)
+ throw new InvalidRecordException("Invalid record size: " + sizeOfBodyInBytes +
+ " exceeds the configured maximum record size of " + maxRecordBodySize + ".");
int totalSizeInBytes = ByteUtils.sizeOfVarint(sizeOfBodyInBytes) + sizeOfBodyInBytes;
return readPartiallyFrom(input, totalSizeInBytes, baseOffset, baseTimestamp,
diff --git a/clients/src/main/java/org/apache/kafka/common/record/internal/DefaultRecordBatch.java b/clients/src/main/java/org/apache/kafka/common/record/internal/DefaultRecordBatch.java
index b20ed9fdcc0..96bc34080c5 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/internal/DefaultRecordBatch.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/internal/DefaultRecordBatch.java
@@ -276,21 +276,21 @@ public class DefaultRecordBatch extends AbstractRecordBatch implements MutableRe
return Compression.of(compressionType()).build().wrapForInput(buffer, magic(), bufferSupplier);
}
- private CloseableIterator<Record> compressedIterator(BufferSupplier bufferSupplier, boolean skipKeyValue) {
+ private CloseableIterator<Record> compressedIterator(BufferSupplier bufferSupplier, boolean skipKeyValue, int maxRecordBodySize) {
final InputStream inputStream = recordInputStream(bufferSupplier);
if (skipKeyValue) {
return new StreamRecordIterator(inputStream) {
@Override
protected Record doReadRecord(long baseOffset, long baseTimestamp, int baseSequence, Long logAppendTime) throws IOException {
- return DefaultRecord.readPartiallyFrom(inputStream, baseOffset, baseTimestamp, baseSequence, logAppendTime);
+ return DefaultRecord.readPartiallyFrom(inputStream, baseOffset, baseTimestamp, baseSequence, logAppendTime, maxRecordBodySize);
}
};
} else {
return new StreamRecordIterator(inputStream) {
@Override
protected Record doReadRecord(long baseOffset, long baseTimestamp, int baseSequence, Long logAppendTime) throws IOException {
- return DefaultRecord.readFrom(inputStream, baseOffset, baseTimestamp, baseSequence, logAppendTime);
+ return DefaultRecord.readFrom(inputStream, baseOffset, baseTimestamp, baseSequence, logAppendTime, maxRecordBodySize);
}
};
}
@@ -328,7 +328,7 @@ public class DefaultRecordBatch extends AbstractRecordBatch implements MutableRe
// for a normal iterator, we cannot ensure that the underlying compression stream is closed,
// so we decompress the full record set here. Use cases which call for a lower memory footprint
// can use `streamingIterator` at the cost of additional complexity
- try (CloseableIterator<Record> iterator = compressedIterator(BufferSupplier.NO_CACHING, false)) {
+ try (CloseableIterator<Record> iterator = compressedIterator(BufferSupplier.NO_CACHING, false, Records.SOFT_MAX_ARRAY_LENGTH)) {
List<Record> records = new ArrayList<>(count());
while (iterator.hasNext())
records.add(iterator.next());
@@ -338,6 +338,11 @@ public class DefaultRecordBatch extends AbstractRecordBatch implements MutableRe
@Override
public CloseableIterator<Record> skipKeyValueIterator(BufferSupplier bufferSupplier) {
+ return skipKeyValueIterator(bufferSupplier, Records.SOFT_MAX_ARRAY_LENGTH);
+ }
+
+ @Override
+ public CloseableIterator<Record> skipKeyValueIterator(BufferSupplier bufferSupplier, int maxRecordBodySize) {
if (count() == 0) {
return CloseableIterator.wrap(Collections.emptyIterator());
}
@@ -352,13 +357,18 @@ public class DefaultRecordBatch extends AbstractRecordBatch implements MutableRe
// we define this to be a closable iterator so that caller (i.e. the log validator) needs to close it
// while we can save memory footprint of not decompressing the full record set ahead of time
- return compressedIterator(bufferSupplier, true);
+ return compressedIterator(bufferSupplier, true, maxRecordBodySize);
}
@Override
public CloseableIterator<Record> streamingIterator(BufferSupplier bufferSupplier) {
+ return streamingIterator(bufferSupplier, Records.SOFT_MAX_ARRAY_LENGTH);
+ }
+
+ @Override
+ public CloseableIterator<Record> streamingIterator(BufferSupplier bufferSupplier, int maxRecordBodySize) {
if (isCompressed())
- return compressedIterator(bufferSupplier, false);
+ return compressedIterator(bufferSupplier, false, maxRecordBodySize);
else
return uncompressedIterator();
}
diff --git a/clients/src/main/java/org/apache/kafka/common/record/internal/FileLogInputStream.java b/clients/src/main/java/org/apache/kafka/common/record/internal/FileLogInputStream.java
index 1b038e18325..771a8c4e0e1 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/internal/FileLogInputStream.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/internal/FileLogInputStream.java
@@ -159,6 +159,11 @@ public class FileLogInputStream implements LogInputStream<FileLogInputStream.Fil
return loadFullBatch().streamingIterator(bufferSupplier);
}
+ @Override
+ public CloseableIterator<Record> streamingIterator(BufferSupplier bufferSupplier, int maxRecordBodySize) {
+ return loadFullBatch().streamingIterator(bufferSupplier, maxRecordBodySize);
+ }
+
@Override
public boolean isValid() {
return loadFullBatch().isValid();
diff --git a/clients/src/main/java/org/apache/kafka/common/record/internal/FileRecords.java b/clients/src/main/java/org/apache/kafka/common/record/internal/FileRecords.java
index 7a7ae0f4e00..ec4580ec033 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/internal/FileRecords.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/internal/FileRecords.java
@@ -21,6 +21,8 @@ import org.apache.kafka.common.network.TransferableChannel;
import org.apache.kafka.common.record.internal.FileLogInputStream.FileChannelRecordBatch;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.common.utils.internals.AbstractIterator;
+import org.apache.kafka.common.utils.internals.BufferSupplier;
+import org.apache.kafka.common.utils.internals.CloseableIterator;
import java.io.Closeable;
import java.io.File;
@@ -351,17 +353,25 @@ public class FileRecords extends AbstractRecords implements Closeable {
* @param targetTimestamp The timestamp to search for.
* @param startingPosition The starting position to search.
* @param startingOffset The starting offset to search.
+ * @param maxRecordBodySize The maximum declared (decompressed) body size of a single record; a compressed
+ * record exceeding it is rejected with an InvalidRecordException before its body is
+ * allocated. Pass {@link Records#SOFT_MAX_ARRAY_LENGTH} for no limit beyond the
+ * array-length ceiling.
* @return The timestamp and offset of the message found. Null if no message is found.
*/
- public TimestampAndOffset searchForTimestamp(long targetTimestamp, int startingPosition, long startingOffset) {
+ public TimestampAndOffset searchForTimestamp(long targetTimestamp, int startingPosition, long startingOffset,
+ int maxRecordBodySize) {
for (RecordBatch batch : batchesFrom(startingPosition)) {
if (batch.maxTimestamp() >= targetTimestamp) {
// We found a message
- for (Record record : batch) {
- long timestamp = record.timestamp();
- if (timestamp >= targetTimestamp && record.offset() >= startingOffset)
- return new TimestampAndOffset(timestamp, record.offset(),
- maybeLeaderEpoch(batch.partitionLeaderEpoch()));
+ try (CloseableIterator<Record> iterator = batch.streamingIterator(BufferSupplier.NO_CACHING, maxRecordBodySize)) {
+ while (iterator.hasNext()) {
+ Record record = iterator.next();
+ long timestamp = record.timestamp();
+ if (timestamp >= targetTimestamp && record.offset() >= startingOffset)
+ return new TimestampAndOffset(timestamp, record.offset(),
+ maybeLeaderEpoch(batch.partitionLeaderEpoch()));
+ }
}
}
}
diff --git a/clients/src/main/java/org/apache/kafka/common/record/internal/MemoryRecords.java b/clients/src/main/java/org/apache/kafka/common/record/internal/MemoryRecords.java
index 08caf0859cc..9d59f67859b 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/internal/MemoryRecords.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/internal/MemoryRecords.java
@@ -127,6 +127,9 @@ public class MemoryRecords extends AbstractRecords {
/**
* Filter the records into the provided ByteBuffer.
*
+ * Note: This method is also used to convert the first timestamp of the batch (which is usually the timestamp of the first record)
+ * to the delete horizon of the tombstones or txn markers which are present in the batch.
+ *
* @param filter The filter function
* @param destinationBuffer The byte buffer to write the filtered records to
* @param decompressionBufferSupplier The supplier of ByteBuffer(s) used for decompression if supported. For small
@@ -134,21 +137,17 @@ public class MemoryRecords extends AbstractRecords {
* dominate the cost of decompressing and iterating over the records in the
* batch. As such, a supplier that reuses buffers will have a significant
* performance impact.
+ * @param maxRecordBodySize The maximum declared (decompressed) body size of a single record; a record
+ * exceeding it is rejected with an {@link org.apache.kafka.common.InvalidRecordException}
+ * before its body is allocated. Pass {@link Records#SOFT_MAX_ARRAY_LENGTH} for
+ * no limit beyond the array-length ceiling.
* @return A FilterResult with a summary of the output (for metrics) and potentially an overflow buffer
*/
- public FilterResult filterTo(RecordFilter filter, ByteBuffer destinationBuffer, BufferSupplier decompressionBufferSupplier) {
- return filterTo(batches(), filter, destinationBuffer, decompressionBufferSupplier);
- }
-
- /**
- * Note: This method is also used to convert the first timestamp of the batch (which is usually the timestamp of the first record)
- * to the delete horizon of the tombstones or txn markers which are present in the batch.
- */
- private static FilterResult filterTo(Iterable<MutableRecordBatch> batches, RecordFilter filter,
- ByteBuffer destinationBuffer, BufferSupplier decompressionBufferSupplier) {
+ public FilterResult filterTo(RecordFilter filter, ByteBuffer destinationBuffer, BufferSupplier decompressionBufferSupplier,
+ int maxRecordBodySize) {
FilterResult filterResult = new FilterResult(destinationBuffer);
ByteBufferOutputStream bufferOutputStream = new ByteBufferOutputStream(destinationBuffer);
- for (MutableRecordBatch batch : batches) {
+ for (MutableRecordBatch batch : batches()) {
final BatchRetentionResult batchRetentionResult = filter.checkBatchRetention(batch);
final boolean containsMarkerForEmptyTxn = batchRetentionResult.containsMarkerForEmptyTxn;
final BatchRetention batchRetention = batchRetentionResult.batchRetention;
@@ -159,7 +158,7 @@ public class MemoryRecords extends AbstractRecords {
continue;
final BatchFilterResult iterationResult = filterBatch(batch, decompressionBufferSupplier, filterResult,
- filter);
+ filter, maxRecordBodySize);
List<Record> retainedRecords = iterationResult.retainedRecords;
boolean containsTombstones = iterationResult.containsTombstones;
boolean writeOriginalBatch = iterationResult.writeOriginalBatch;
@@ -216,8 +215,9 @@ public class MemoryRecords extends AbstractRecords {
private static BatchFilterResult filterBatch(RecordBatch batch,
BufferSupplier decompressionBufferSupplier,
FilterResult filterResult,
- RecordFilter filter) {
- try (final CloseableIterator<Record> iterator = batch.streamingIterator(decompressionBufferSupplier)) {
+ RecordFilter filter,
+ int maxRecordBodySize) {
+ try (final CloseableIterator<Record> iterator = batch.streamingIterator(decompressionBufferSupplier, maxRecordBodySize)) {
long maxOffset = -1;
boolean containsTombstones = false;
// Convert records with old record versions
diff --git a/clients/src/main/java/org/apache/kafka/common/record/internal/MutableRecordBatch.java b/clients/src/main/java/org/apache/kafka/common/record/internal/MutableRecordBatch.java
index 6c7c7bf695b..5ea7883e118 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/internal/MutableRecordBatch.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/internal/MutableRecordBatch.java
@@ -66,4 +66,13 @@ public interface MutableRecordBatch extends RecordBatch {
* @return The closeable iterator
*/
CloseableIterator<Record> skipKeyValueIterator(BufferSupplier bufferSupplier);
+
+ /**
+ * Variant of {@link #skipKeyValueIterator(BufferSupplier)} that rejects any record whose declared
+ * (decompressed) body size exceeds {@code maxRecordBodySize}; see
+ * {@link RecordBatch#streamingIterator(BufferSupplier, int)}.
+ *
+ * @return The closeable iterator
+ */
+ CloseableIterator<Record> skipKeyValueIterator(BufferSupplier bufferSupplier, int maxRecordBodySize);
}
diff --git a/clients/src/main/java/org/apache/kafka/common/record/internal/RecordBatch.java b/clients/src/main/java/org/apache/kafka/common/record/internal/RecordBatch.java
index 16bf94133cf..7f37f72bf0f 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/internal/RecordBatch.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/internal/RecordBatch.java
@@ -240,6 +240,15 @@ public interface RecordBatch extends Iterable<Record> {
*/
CloseableIterator<Record> streamingIterator(BufferSupplier decompressionBufferSupplier);
+ /**
+ * Variant of {@link #streamingIterator(BufferSupplier)} that rejects any record whose declared
+ * (decompressed) body size exceeds {@code maxRecordBodySize} before allocating the body buffer.
+ * The single-argument overload passes {@link Records#SOFT_MAX_ARRAY_LENGTH} (no effective limit).
+ *
+ * @return The closeable iterator
+ */
+ CloseableIterator<Record> streamingIterator(BufferSupplier decompressionBufferSupplier, int maxRecordBodySize);
+
/**
* Check whether this is a control batch (i.e. whether the control bit is set in the batch attributes).
* For magic versions prior to 2, this is always false.
@@ -253,12 +262,15 @@ public interface RecordBatch extends Iterable<Record> {
* noted:
* 1) that the earliest offset will return if there are multi records having same (max) timestamp
* 2) it always returns None if the {@link RecordBatch#magic()} is equal to {@link RecordBatch#MAGIC_VALUE_V0}
+ * @param maxRecordBodySize The maximum declared (decompressed) body size of a single record; a compressed record
+ * exceeding it is rejected with an InvalidRecordException before its body is allocated.
+ * Pass {@link Records#SOFT_MAX_ARRAY_LENGTH} for no limit beyond the array-length ceiling.
* @return offset of max timestamp
*/
- default Optional<Long> offsetOfMaxTimestamp() {
+ default Optional<Long> offsetOfMaxTimestamp(int maxRecordBodySize) {
if (magic() == RecordBatch.MAGIC_VALUE_V0) return Optional.empty();
long maxTimestamp = maxTimestamp();
- try (CloseableIterator<Record> iter = streamingIterator(BufferSupplier.create())) {
+ try (CloseableIterator<Record> iter = streamingIterator(BufferSupplier.create(), maxRecordBodySize)) {
while (iter.hasNext()) {
Record record = iter.next();
if (maxTimestamp == record.timestamp()) return Optional.of(record.offset());
diff --git a/clients/src/main/java/org/apache/kafka/common/record/internal/Records.java b/clients/src/main/java/org/apache/kafka/common/record/internal/Records.java
index 3264698b55a..d42c8c7a766 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/internal/Records.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/internal/Records.java
@@ -54,6 +54,13 @@ public interface Records extends TransferableRecords {
int MAGIC_LENGTH = 1;
int HEADER_SIZE_UP_TO_MAGIC = MAGIC_OFFSET + MAGIC_LENGTH;
+ // The largest byte array the JVM reliably allocates, mirroring the JDK's internal
+ // ArraysSupport.SOFT_MAX_ARRAY_LENGTH: some VMs reserve header words inside the array object,
+ // so allocations at Integer.MAX_VALUE may fail with OutOfMemoryError regardless of available
+ // heap. Used as the ceiling (and default) for the declared size of a record body read from a
+ // decompressed stream, where the size must be checked before it is allocated.
+ int SOFT_MAX_ARRAY_LENGTH = Integer.MAX_VALUE - 8;
+
/**
* Get the record batches. Note that the signature allows subclasses
* to return a more specific batch type. This enables optimizations such as in-place offset
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchRequestManagerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchRequestManagerTest.java
index 85b34a3e48d..043255e4947 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchRequestManagerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchRequestManagerTest.java
@@ -2795,7 +2795,7 @@ public class FetchRequestManagerTest {
protected boolean shouldRetainRecord(RecordBatch recordBatch, Record record) {
return record.key() != null;
}
- }, ByteBuffer.allocate(1024), BufferSupplier.NO_CACHING);
+ }, ByteBuffer.allocate(1024), BufferSupplier.NO_CACHING, Records.SOFT_MAX_ARRAY_LENGTH);
result.outputBuffer().flip();
MemoryRecords compactedRecords = MemoryRecords.readableRecords(result.outputBuffer());
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
index fe1ecca288e..85ce007f43a 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
@@ -2551,7 +2551,7 @@ public class FetcherTest {
protected boolean shouldRetainRecord(RecordBatch recordBatch, Record record) {
return record.key() != null;
}
- }, ByteBuffer.allocate(1024), BufferSupplier.NO_CACHING);
+ }, ByteBuffer.allocate(1024), BufferSupplier.NO_CACHING, Records.SOFT_MAX_ARRAY_LENGTH);
result.outputBuffer().flip();
MemoryRecords compactedRecords = MemoryRecords.readableRecords(result.outputBuffer());
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManagerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManagerTest.java
index 76e71f86837..6d89cc246c1 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManagerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManagerTest.java
@@ -61,6 +61,7 @@ import org.apache.kafka.common.record.internal.MemoryRecords;
import org.apache.kafka.common.record.internal.MemoryRecordsBuilder;
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.requests.MetadataResponse;
import org.apache.kafka.common.requests.RequestHeader;
@@ -2060,7 +2061,7 @@ public class ShareConsumeRequestManagerTest {
protected boolean shouldRetainRecord(RecordBatch recordBatch, Record record) {
return record.key() != null;
}
- }, ByteBuffer.allocate(1024), BufferSupplier.NO_CACHING);
+ }, ByteBuffer.allocate(1024), BufferSupplier.NO_CACHING, Records.SOFT_MAX_ARRAY_LENGTH);
result.outputBuffer().flip();
MemoryRecords compactedRecords = MemoryRecords.readableRecords(result.outputBuffer());
diff --git a/clients/src/test/java/org/apache/kafka/common/record/internal/AbstractLegacyRecordBatchTest.java b/clients/src/test/java/org/apache/kafka/common/record/internal/AbstractLegacyRecordBatchTest.java
index 242896e0fb8..448b8372184 100644
--- a/clients/src/test/java/org/apache/kafka/common/record/internal/AbstractLegacyRecordBatchTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/record/internal/AbstractLegacyRecordBatchTest.java
@@ -21,9 +21,15 @@ import org.apache.kafka.common.compress.Compression;
import org.apache.kafka.common.record.TimestampType;
import org.apache.kafka.common.record.internal.AbstractLegacyRecordBatch.ByteBufferLegacyRecordBatch;
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.CloseableIterator;
import org.junit.jupiter.api.Test;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.List;
@@ -251,4 +257,63 @@ public class AbstractLegacyRecordBatchTest {
}
}
+ /**
+ * A compressed legacy (v0/v1) wrapper whose decompressed payload declares a single inner record
+ * with a forged size and no payload bytes. The deep-decode path must reject the declared size
+ * before allocating it.
+ */
+ private static ByteBufferLegacyRecordBatch poisonedCompressedLegacyBatch(byte magic, int forgedInnerSize) throws IOException {
+ ByteBufferOutputStream innerOut = new ByteBufferOutputStream(64);
+ try (DataOutputStream compressed = new DataOutputStream(
+ Compression.gzip().build().wrapForOutput(innerOut, magic))) {
+ compressed.writeLong(0L); // inner offset
+ compressed.writeInt(forgedInnerSize); // forged inner record size, no payload follows
+ }
+ ByteBuffer inner = innerOut.buffer();
+ inner.flip();
+ byte[] compressedBytes = new byte[inner.remaining()];
+ inner.get(compressedBytes);
+
+ LegacyRecord wrapper = LegacyRecord.create(magic, 1L, null, compressedBytes,
+ CompressionType.GZIP, TimestampType.CREATE_TIME);
+ ByteBuffer buffer = ByteBuffer.allocate(Records.LOG_OVERHEAD + wrapper.sizeInBytes());
+ AbstractLegacyRecordBatch.writeHeader(buffer, 0L, wrapper.sizeInBytes());
+ buffer.put(wrapper.buffer().duplicate());
+ buffer.flip();
+ return new ByteBufferLegacyRecordBatch(buffer);
+ }
+
+ // The compressed deep-decode path constructs its inner stream with maxMessageSize =
+ // Integer.MAX_VALUE, so only the per-record guard stands between a forged inner size and
+ // ByteBuffer.allocate. SOFT_MAX_ARRAY_LENGTH + 1 pins the default (array-length) threshold
+ // without attempting a ~2 GiB allocation.
+ @Test
+ public void testCompressedDeepDecodeRejectsForgedInnerSizeExceedingArrayLengthLimit() throws IOException {
+ ByteBufferLegacyRecordBatch batch = poisonedCompressedLegacyBatch(
+ RecordBatch.MAGIC_VALUE_V1, Records.SOFT_MAX_ARRAY_LENGTH + 1);
+ // the deep iterator decodes eagerly in its constructor
+ InvalidRecordException ex = assertThrows(InvalidRecordException.class,
+ () -> batch.streamingIterator(BufferSupplier.NO_CACHING));
+ assertTrue(ex.getMessage().contains("exceeds the configured maximum record size"),
+ "expected the pre-allocation record-size guard, got: " + ex.getMessage());
+ }
+
+ // A genuine compressed legacy record whose inner size exceeds the configured limit is rejected
+ // before allocation, while a generous limit decodes it.
+ @Test
+ public void testCompressedDeepDecodeEnforcesConfiguredMaxRecordBodySize() {
+ MemoryRecords records = MemoryRecords.withRecords(RecordBatch.MAGIC_VALUE_V1, 0L,
+ Compression.gzip().build(), TimestampType.CREATE_TIME,
+ new SimpleRecord(1L, "key".getBytes(), new byte[1000]));
+ ByteBufferLegacyRecordBatch batch = new ByteBufferLegacyRecordBatch(records.buffer());
+
+ try (CloseableIterator<Record> iterator = batch.streamingIterator(BufferSupplier.NO_CACHING, 10_000)) {
+ assertTrue(iterator.hasNext());
+ }
+ InvalidRecordException ex = assertThrows(InvalidRecordException.class,
+ () -> batch.streamingIterator(BufferSupplier.NO_CACHING, 100));
+ assertTrue(ex.getMessage().contains("exceeds the configured maximum record size"),
+ "expected the configured-maximum guard, got: " + ex.getMessage());
+ }
+
}
diff --git a/clients/src/test/java/org/apache/kafka/common/record/internal/DefaultRecordBatchTest.java b/clients/src/test/java/org/apache/kafka/common/record/internal/DefaultRecordBatchTest.java
index f48168a26a7..dca3db18553 100644
--- a/clients/src/test/java/org/apache/kafka/common/record/internal/DefaultRecordBatchTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/record/internal/DefaultRecordBatchTest.java
@@ -25,6 +25,8 @@ import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.common.record.TimestampType;
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.common.utils.internals.ChunkedBytesStream;
import org.apache.kafka.common.utils.internals.CloseableIterator;
import org.apache.kafka.test.TestUtils;
@@ -35,11 +37,13 @@ import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.EnumSource;
import org.junit.jupiter.params.provider.MethodSource;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.List;
+import java.util.Optional;
import java.util.Random;
import java.util.stream.Stream;
@@ -567,6 +571,97 @@ public class DefaultRecordBatchTest {
assertEquals(Integer.MAX_VALUE, DefaultRecordBatch.decrementSequence(0, 1));
}
+ // =============================================================================================
+ // Per-record decompressed-body-size guard: the compressed iterators reject a record whose
+ // declared (decompressed) body exceeds the given limit — Records.SOFT_MAX_ARRAY_LENGTH for the
+ // limit-less overloads — with InvalidRecordException, before allocating the body buffer.
+ // =============================================================================================
+
+ /**
+ * A compressed V2 batch whose records section declares a single record with a forged body size
+ * and no body bytes; the guard must fire on the declared size before any allocation is attempted.
+ */
+ private static DefaultRecordBatch poisonedCompressedV2Batch(Compression compression, int forgedBodySize) throws IOException {
+ ByteBufferOutputStream payloadOut = new ByteBufferOutputStream(64);
+ try (DataOutputStream compressed = new DataOutputStream(
+ compression.wrapForOutput(payloadOut, RecordBatch.MAGIC_VALUE_V2))) {
+ ByteUtils.writeVarint(forgedBodySize, compressed);
+ }
+ ByteBuffer payload = payloadOut.buffer();
+ payload.flip();
+
+ int sizeInBytes = DefaultRecordBatch.RECORD_BATCH_OVERHEAD + payload.remaining();
+ ByteBuffer buffer = ByteBuffer.allocate(sizeInBytes);
+ buffer.position(DefaultRecordBatch.RECORD_BATCH_OVERHEAD);
+ buffer.put(payload);
+ buffer.position(0);
+ DefaultRecordBatch.writeHeader(buffer, 0L, 0, sizeInBytes, RecordBatch.MAGIC_VALUE_V2,
+ compression.type(), TimestampType.CREATE_TIME, 0L, 0L, RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_EPOCH, RecordBatch.NO_SEQUENCE, false, false, false, 0, 1);
+ buffer.position(0);
+ return new DefaultRecordBatch(buffer);
+ }
+
+ // A forged body size above the array-length ceiling is rejected by the limit-less iterator
+ // (which applies Records.SOFT_MAX_ARRAY_LENGTH) instead of reaching ByteBuffer.allocate and
+ // raising OutOfMemoryError. SOFT_MAX_ARRAY_LENGTH + 1 pins the exact threshold without
+ // attempting a ~2 GiB allocation.
+ @Test
+ public void testStreamingIteratorRejectsForgedBodySizeExceedingArrayLengthLimit() throws IOException {
+ DefaultRecordBatch poison = poisonedCompressedV2Batch(Compression.gzip().build(), Records.SOFT_MAX_ARRAY_LENGTH + 1);
+ try (CloseableIterator<Record> iterator = poison.streamingIterator(BufferSupplier.NO_CACHING)) {
+ InvalidRecordException ex = assertThrows(InvalidRecordException.class, iterator::next);
+ assertTrue(ex.getMessage().contains("exceeds the configured maximum record size"),
+ "expected the pre-allocation record-size guard, got: " + ex.getMessage());
+ }
+ }
+
+ // A genuine record whose decompressed body exceeds the configured limit is rejected before
+ // allocation, while a generous limit accepts it — on both compressed iterator variants.
+ @Test
+ public void testCompressedIteratorsEnforceConfiguredMaxRecordBodySize() {
+ DefaultRecordBatch batch = recordBatchWithValueSize(1000);
+
+ try (CloseableIterator<Record> iterator = batch.streamingIterator(BufferSupplier.NO_CACHING, 10_000)) {
+ assertNotNull(iterator.next());
+ }
+ try (CloseableIterator<Record> iterator = batch.streamingIterator(BufferSupplier.NO_CACHING, 100)) {
+ InvalidRecordException ex = assertThrows(InvalidRecordException.class, iterator::next);
+ assertTrue(ex.getMessage().contains("exceeds the configured maximum record size"),
+ "expected the configured-maximum guard, got: " + ex.getMessage());
+ }
+
+ try (CloseableIterator<Record> iterator = batch.skipKeyValueIterator(BufferSupplier.NO_CACHING, 10_000)) {
+ assertNotNull(iterator.next());
+ }
+ try (CloseableIterator<Record> iterator = batch.skipKeyValueIterator(BufferSupplier.NO_CACHING, 100)) {
+ InvalidRecordException ex = assertThrows(InvalidRecordException.class, iterator::next);
+ assertTrue(ex.getMessage().contains("exceeds the configured maximum record size"),
+ "expected the configured-maximum guard, got: " + ex.getMessage());
+ }
+ }
+
+ // offsetOfMaxTimestamp decompresses the batch on the broker when it resolves ListOffsets MAX_TIMESTAMP to an
+ // exact offset, so it honours the same per-record limit as the compressed iterators.
+ @Test
+ public void testOffsetOfMaxTimestampEnforcesConfiguredMaxRecordBodySize() {
+ DefaultRecordBatch batch = recordBatchWithValueSize(1000);
+
+ assertEquals(Optional.of(0L), batch.offsetOfMaxTimestamp(10_000));
+
+ InvalidRecordException ex = assertThrows(InvalidRecordException.class, () -> batch.offsetOfMaxTimestamp(100));
+ assertTrue(ex.getMessage().contains("exceeds the configured maximum record size"),
+ "expected the configured-maximum guard, got: " + ex.getMessage());
+ }
+
+ private static DefaultRecordBatch recordBatchWithValueSize(int valueSize) {
+ ByteBuffer buf = ByteBuffer.allocate(2048);
+ MemoryRecordsBuilder builder = MemoryRecords.builder(buf, RecordBatch.MAGIC_VALUE_V2,
+ Compression.gzip().build(), TimestampType.CREATE_TIME, 0L);
+ builder.appendWithOffset(0, 12L, "key".getBytes(), new byte[valueSize]);
+ return new DefaultRecordBatch(builder.build().buffer());
+ }
+
private static DefaultRecordBatch recordsWithInvalidRecordCount(Byte magicValue, long timestamp,
CompressionType codec, int invalidCount) {
ByteBuffer buf = ByteBuffer.allocate(512);
diff --git a/clients/src/test/java/org/apache/kafka/common/record/internal/DefaultRecordTest.java b/clients/src/test/java/org/apache/kafka/common/record/internal/DefaultRecordTest.java
index ea8e15701f5..d9e133b1066 100644
--- a/clients/src/test/java/org/apache/kafka/common/record/internal/DefaultRecordTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/record/internal/DefaultRecordTest.java
@@ -34,6 +34,7 @@ import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class DefaultRecordTest {
@Test
@@ -103,7 +104,7 @@ public class DefaultRecordTest {
// test for input stream input
try (ByteBufferInputStream inpStream = new ByteBufferInputStream(buffer.asReadOnlyBuffer())) {
assertThrows(InvalidRecordException.class,
- () -> DefaultRecord.readFrom(inpStream, baseOffset, baseTimestamp, baseSequence, null));
+ () -> DefaultRecord.readFrom(inpStream, baseOffset, baseTimestamp, baseSequence, null, Records.SOFT_MAX_ARRAY_LENGTH));
}
// test for buffer input
assertThrows(InvalidRecordException.class,
@@ -469,7 +470,7 @@ public class DefaultRecordTest {
// test for input stream input
try (ByteBufferInputStream inpStream = new ByteBufferInputStream(buffer.asReadOnlyBuffer())) {
- DefaultRecord record = DefaultRecord.readFrom(inpStream, baseOffset, baseTimestamp, RecordBatch.NO_SEQUENCE, null);
+ DefaultRecord record = DefaultRecord.readFrom(inpStream, baseOffset, baseTimestamp, RecordBatch.NO_SEQUENCE, null, Records.SOFT_MAX_ARRAY_LENGTH);
assertNotNull(record);
assertEquals(RecordBatch.NO_SEQUENCE, record.sequence());
}
@@ -491,10 +492,116 @@ public class DefaultRecordTest {
assertDecodingRecordFromBufferThrowsInvalidRecordException(buf);
}
+ // =============================================================================================
+ // Configurable per-record decompressed-body-size limit (max.decompressed.message.bytes).
+ // Broker decode paths pass the configured limit to the InputStream decoders; a record whose
+ // declared (decompressed) body exceeds it is rejected with InvalidRecordException BEFORE
+ // allocation, so the limit bounds the allocation. Callers that do not thread a limit use the
+ // overloads without one, which apply Records.SOFT_MAX_ARRAY_LENGTH — the array-length ceiling.
+ // =============================================================================================
+
+ private static byte[] recordWithForgedBodySize(int declaredBodySize) {
+ // Only the leading size varint matters: the guard fires before any body bytes are read.
+ ByteBuffer buf = ByteBuffer.allocate(16);
+ ByteUtils.writeVarint(declaredBodySize, buf);
+ buf.put((byte) 0); // attribute byte, never reached when the guard fires
+ buf.flip();
+ byte[] bytes = new byte[buf.remaining()];
+ buf.get(bytes);
+ return bytes;
+ }
+
+ // A negative size is forgeable via the zig-zag varint; sizes above the array length limit can
+ // never be allocated. Both are rejected before any allocation. SOFT_MAX_ARRAY_LENGTH + 1 pins
+ // the exact upper threshold without attempting a ~2 GiB allocation.
+ @Test
+ public void testReadFromStreamRejectsInvalidBodySize() throws IOException {
+ assertReadFromStreamRejectsBodySize(-1, "is negative");
+ assertReadFromStreamRejectsBodySize(Integer.MAX_VALUE, "exceeds the configured maximum record size");
+ assertReadFromStreamRejectsBodySize(Records.SOFT_MAX_ARRAY_LENGTH + 1, "exceeds the configured maximum record size");
+ }
+
+ private static void assertReadFromStreamRejectsBodySize(int declaredBodySize, String expectedMessage) throws IOException {
+ byte[] rec = recordWithForgedBodySize(declaredBodySize);
+ try (InputStream in = new ByteBufferInputStream(ByteBuffer.wrap(rec))) {
+ InvalidRecordException ex = assertThrows(InvalidRecordException.class,
+ () -> DefaultRecord.readFrom(in, 0L, 0L, RecordBatch.NO_SEQUENCE, null, Records.SOFT_MAX_ARRAY_LENGTH));
+ assertTrue(ex.getMessage().contains(expectedMessage),
+ "expected '" + expectedMessage + "', got: " + ex.getMessage());
+ }
+ }
+
+ @Test
+ public void testReadPartiallyFromStreamRejectsInvalidBodySize() throws IOException {
+ assertReadPartiallyFromStreamRejectsBodySize(-1, "is negative");
+ assertReadPartiallyFromStreamRejectsBodySize(Integer.MAX_VALUE, "exceeds the configured maximum record size");
+ assertReadPartiallyFromStreamRejectsBodySize(Records.SOFT_MAX_ARRAY_LENGTH + 1, "exceeds the configured maximum record size");
+ }
+
+ private static void assertReadPartiallyFromStreamRejectsBodySize(int declaredBodySize, String expectedMessage) throws IOException {
+ byte[] rec = recordWithForgedBodySize(declaredBodySize);
+ try (InputStream in = new ByteBufferInputStream(ByteBuffer.wrap(rec))) {
+ InvalidRecordException ex = assertThrows(InvalidRecordException.class,
+ () -> DefaultRecord.readPartiallyFrom(in, 0L, 0L, RecordBatch.NO_SEQUENCE, null, Records.SOFT_MAX_ARRAY_LENGTH));
+ assertTrue(ex.getMessage().contains(expectedMessage),
+ "expected '" + expectedMessage + "', got: " + ex.getMessage());
+ }
+ }
+
+ // The forged body size (1000) is well under the array-length limit but over the configured max
+ // (100), so the configurable guard fires (not the array-length guard) before any allocation.
+ @Test
+ public void testReadFromStreamRejectsBodySizeExceedingConfiguredMax() throws IOException {
+ byte[] rec = recordWithForgedBodySize(1000);
+ try (InputStream in = new ByteBufferInputStream(ByteBuffer.wrap(rec))) {
+ InvalidRecordException ex = assertThrows(InvalidRecordException.class,
+ () -> DefaultRecord.readFrom(in, 0L, 0L, RecordBatch.NO_SEQUENCE, null, 100));
+ assertTrue(ex.getMessage().contains("exceeds the configured maximum record size"),
+ "expected the configured-maximum guard, got: " + ex.getMessage());
+ }
+ }
+
+ @Test
+ public void testReadPartiallyFromStreamRejectsBodySizeExceedingConfiguredMax() throws IOException {
+ byte[] rec = recordWithForgedBodySize(1000);
+ try (InputStream in = new ByteBufferInputStream(ByteBuffer.wrap(rec))) {
+ InvalidRecordException ex = assertThrows(InvalidRecordException.class,
+ () -> DefaultRecord.readPartiallyFrom(in, 0L, 0L, RecordBatch.NO_SEQUENCE, null, 100));
+ assertTrue(ex.getMessage().contains("exceeds the configured maximum record size"),
+ "expected the configured-maximum guard, got: " + ex.getMessage());
+ }
+ }
+
+ // A genuinely-serialized record decodes when the configured limit is generous, and is rejected
+ // (before allocation) when the limit is below its body, proving the guard enforces the limit on
+ // real records without rejecting valid ones.
+ @Test
+ public void testReadFromStreamWithConfiguredMaxAcceptsValidRecordAndRejectsWhenTooSmall() throws IOException {
+ ByteBuffer key = ByteBuffer.wrap("hi".getBytes());
+ ByteBuffer value = ByteBuffer.wrap("there".getBytes());
+ ByteBufferOutputStream out = new ByteBufferOutputStream(1024);
+ DefaultRecord.writeTo(new DataOutputStream(out), 0, 0L, key, value, new Header[0]);
+ ByteBuffer buffer = out.buffer();
+ buffer.flip();
+
+ // A generous limit accepts the valid record.
+ try (InputStream in = new ByteBufferInputStream(buffer.duplicate())) {
+ DefaultRecord record = DefaultRecord.readFrom(in, 0L, 0L, RecordBatch.NO_SEQUENCE, null, 1024);
+ assertNotNull(record);
+ }
+ // A 1-byte limit is below the real body, so the record is rejected before allocation.
+ try (InputStream in = new ByteBufferInputStream(buffer.duplicate())) {
+ InvalidRecordException ex = assertThrows(InvalidRecordException.class,
+ () -> DefaultRecord.readFrom(in, 0L, 0L, RecordBatch.NO_SEQUENCE, null, 1));
+ assertTrue(ex.getMessage().contains("exceeds the configured maximum record size"),
+ "expected the configured-maximum guard, got: " + ex.getMessage());
+ }
+ }
+
private static void assertPartiallyDecodingRecordsFromBufferThrowsInvalidRecordException(ByteBuffer buf) throws IOException {
try (InputStream inputStream = new ByteBufferInputStream(buf)) {
assertThrows(InvalidRecordException.class,
- () -> DefaultRecord.readPartiallyFrom(inputStream, 0L, 0L, RecordBatch.NO_SEQUENCE, null));
+ () -> DefaultRecord.readPartiallyFrom(inputStream, 0L, 0L, RecordBatch.NO_SEQUENCE, null, Records.SOFT_MAX_ARRAY_LENGTH));
}
}
@@ -502,7 +609,7 @@ public class DefaultRecordTest {
// test for input stream input
try (ByteBufferInputStream inpStream = new ByteBufferInputStream(buf.asReadOnlyBuffer())) {
assertThrows(InvalidRecordException.class,
- () -> DefaultRecord.readFrom(inpStream, 0L, 0L, RecordBatch.NO_SEQUENCE, null));
+ () -> DefaultRecord.readFrom(inpStream, 0L, 0L, RecordBatch.NO_SEQUENCE, null, Records.SOFT_MAX_ARRAY_LENGTH));
}
// test for buffer input
assertThrows(InvalidRecordException.class,
diff --git a/clients/src/test/java/org/apache/kafka/common/record/internal/FileRecordsTest.java b/clients/src/test/java/org/apache/kafka/common/record/internal/FileRecordsTest.java
index f0831f556d9..9082db17195 100644
--- a/clients/src/test/java/org/apache/kafka/common/record/internal/FileRecordsTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/record/internal/FileRecordsTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.common.record.internal;
+import org.apache.kafka.common.InvalidRecordException;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.compress.Compression;
import org.apache.kafka.common.header.Header;
@@ -458,12 +459,12 @@ public class FileRecordsTest {
appendWithOffsetAndTimestamp(fileRecords, version, 11L, 6, 1);
assertFoundTimestamp(new FileRecords.TimestampAndOffset(10L, 5, Optional.of(0)),
- fileRecords.searchForTimestamp(9L, 0, 0L), version);
+ fileRecords.searchForTimestamp(9L, 0, 0L, Records.SOFT_MAX_ARRAY_LENGTH), version);
assertFoundTimestamp(new FileRecords.TimestampAndOffset(10L, 5, Optional.of(0)),
- fileRecords.searchForTimestamp(10L, 0, 0L), version);
+ fileRecords.searchForTimestamp(10L, 0, 0L, Records.SOFT_MAX_ARRAY_LENGTH), version);
assertFoundTimestamp(new FileRecords.TimestampAndOffset(11L, 6, Optional.of(1)),
- fileRecords.searchForTimestamp(11L, 0, 0L), version);
- assertNull(fileRecords.searchForTimestamp(12L, 0, 0L));
+ fileRecords.searchForTimestamp(11L, 0, 0L, Records.SOFT_MAX_ARRAY_LENGTH), version);
+ assertNull(fileRecords.searchForTimestamp(12L, 0, 0L, Records.SOFT_MAX_ARRAY_LENGTH));
}
private void assertFoundTimestamp(FileRecords.TimestampAndOffset expected,
@@ -790,4 +791,23 @@ public class FileRecordsTest {
}
fileRecords.flush();
}
+
+
+ @Test
+ public void testSearchForTimestampRejectsCompressedRecordExceedingMaxRecordBodySize() throws IOException {
+ FileRecords fileRecords = FileRecords.open(tempFile(), false, 1024 * 1024, true);
+ long timestamp = 10L;
+ fileRecords.append(MemoryRecords.withRecords(5L, Compression.gzip().build(), 0,
+ new SimpleRecord(timestamp, "key".getBytes(), new byte[1000])));
+ fileRecords.flush();
+
+ // With no configured limit the record is found
+ assertEquals(new FileRecords.TimestampAndOffset(timestamp, 5L, Optional.of(0)),
+ fileRecords.searchForTimestamp(timestamp, 0, 0L, Records.SOFT_MAX_ARRAY_LENGTH));
+
+ // With a limit below the record's decompressed body the lookup is rejected before the body is allocated
+ InvalidRecordException e = assertThrows(InvalidRecordException.class,
+ () -> fileRecords.searchForTimestamp(timestamp, 0, 0L, 100));
+ assertTrue(e.getMessage().contains("exceeds the configured maximum record size of 100"), e.getMessage());
+ }
}
diff --git a/clients/src/test/java/org/apache/kafka/common/record/internal/MemoryRecordsTest.java b/clients/src/test/java/org/apache/kafka/common/record/internal/MemoryRecordsTest.java
index 83b174cccd6..d53c57793ba 100644
--- a/clients/src/test/java/org/apache/kafka/common/record/internal/MemoryRecordsTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/record/internal/MemoryRecordsTest.java
@@ -311,7 +311,7 @@ public class MemoryRecordsTest {
builder.append(12L, null, "c".getBytes());
ByteBuffer filtered = ByteBuffer.allocate(2048);
- builder.build().filterTo(new RetainNonNullKeysFilter(), filtered, BufferSupplier.NO_CACHING);
+ builder.build().filterTo(new RetainNonNullKeysFilter(), filtered, BufferSupplier.NO_CACHING, Records.SOFT_MAX_ARRAY_LENGTH);
filtered.flip();
MemoryRecords filteredRecords = MemoryRecords.readableRecords(filtered);
@@ -364,7 +364,7 @@ public class MemoryRecordsTest {
// delete the records
return false;
}
- }, filtered, BufferSupplier.NO_CACHING);
+ }, filtered, BufferSupplier.NO_CACHING, Records.SOFT_MAX_ARRAY_LENGTH);
// Verify filter result
assertEquals(numRecords, filterResult.messagesRead());
@@ -425,7 +425,7 @@ public class MemoryRecordsTest {
protected boolean shouldRetainRecord(RecordBatch recordBatch, Record record) {
return false;
}
- }, filtered, BufferSupplier.NO_CACHING);
+ }, filtered, BufferSupplier.NO_CACHING, Records.SOFT_MAX_ARRAY_LENGTH);
// Verify filter result
assertEquals(0, filterResult.messagesRead());
@@ -472,7 +472,7 @@ public class MemoryRecordsTest {
protected boolean shouldRetainRecord(RecordBatch recordBatch, Record record) {
return false;
}
- }, filtered, BufferSupplier.NO_CACHING);
+ }, filtered, BufferSupplier.NO_CACHING, Records.SOFT_MAX_ARRAY_LENGTH);
// Verify filter result
assertEquals(0, filterResult.outputBuffer().position());
@@ -548,7 +548,7 @@ public class MemoryRecordsTest {
return new BatchRetentionResult(BatchRetention.RETAIN_EMPTY, false);
}
};
- builder.build().filterTo(recordFilter, filtered, BufferSupplier.NO_CACHING);
+ builder.build().filterTo(recordFilter, filtered, BufferSupplier.NO_CACHING, Records.SOFT_MAX_ARRAY_LENGTH);
filtered.flip();
MemoryRecords filteredRecords = MemoryRecords.readableRecords(filtered);
@@ -650,7 +650,7 @@ public class MemoryRecordsTest {
protected boolean shouldRetainRecord(RecordBatch recordBatch, Record record) {
return true;
}
- }, filtered, BufferSupplier.NO_CACHING);
+ }, filtered, BufferSupplier.NO_CACHING, Records.SOFT_MAX_ARRAY_LENGTH);
filtered.flip();
MemoryRecords filteredRecords = MemoryRecords.readableRecords(filtered);
@@ -686,7 +686,7 @@ public class MemoryRecordsTest {
buffer.flip();
ByteBuffer filtered = ByteBuffer.allocate(2048);
- MemoryRecords.readableRecords(buffer).filterTo(new RetainNonNullKeysFilter(), filtered, BufferSupplier.NO_CACHING);
+ MemoryRecords.readableRecords(buffer).filterTo(new RetainNonNullKeysFilter(), filtered, BufferSupplier.NO_CACHING, Records.SOFT_MAX_ARRAY_LENGTH);
filtered.flip();
MemoryRecords filteredRecords = MemoryRecords.readableRecords(filtered);
@@ -761,7 +761,7 @@ public class MemoryRecordsTest {
buffer.flip();
ByteBuffer filtered = ByteBuffer.allocate(2048);
- MemoryRecords.readableRecords(buffer).filterTo(new RetainNonNullKeysFilter(), filtered, BufferSupplier.NO_CACHING);
+ MemoryRecords.readableRecords(buffer).filterTo(new RetainNonNullKeysFilter(), filtered, BufferSupplier.NO_CACHING, Records.SOFT_MAX_ARRAY_LENGTH);
filtered.flip();
MemoryRecords filteredRecords = MemoryRecords.readableRecords(filtered);
@@ -853,7 +853,7 @@ public class MemoryRecordsTest {
output.rewind();
MemoryRecords.FilterResult result = MemoryRecords.readableRecords(buffer).filterTo(
- new RetainNonNullKeysFilter(), output, BufferSupplier.NO_CACHING);
+ new RetainNonNullKeysFilter(), output, BufferSupplier.NO_CACHING, Records.SOFT_MAX_ARRAY_LENGTH);
buffer.position(buffer.position() + result.bytesRead());
result.outputBuffer().flip();
@@ -900,7 +900,7 @@ public class MemoryRecordsTest {
ByteBuffer filtered = ByteBuffer.allocate(2048);
MemoryRecords.FilterResult result = MemoryRecords.readableRecords(buffer).filterTo(
- new RetainNonNullKeysFilter(), filtered, BufferSupplier.NO_CACHING);
+ new RetainNonNullKeysFilter(), filtered, BufferSupplier.NO_CACHING, Records.SOFT_MAX_ARRAY_LENGTH);
filtered.flip();
@@ -1018,7 +1018,7 @@ public class MemoryRecordsTest {
buffer.flip();
ByteBuffer filtered = ByteBuffer.allocate(2048);
- MemoryRecords.readableRecords(buffer).filterTo(new RetainNonNullKeysFilter(), filtered, BufferSupplier.NO_CACHING);
+ MemoryRecords.readableRecords(buffer).filterTo(new RetainNonNullKeysFilter(), filtered, BufferSupplier.NO_CACHING, Records.SOFT_MAX_ARRAY_LENGTH);
filtered.flip();
MemoryRecords filteredRecords = MemoryRecords.readableRecords(filtered);
diff --git a/core/src/main/scala/kafka/server/ConfigHelper.scala b/core/src/main/scala/kafka/server/ConfigHelper.scala
index a1223218514..22937a393ed 100644
--- a/core/src/main/scala/kafka/server/ConfigHelper.scala
+++ b/core/src/main/scala/kafka/server/ConfigHelper.scala
@@ -91,7 +91,8 @@ class ConfigHelper(metadataCache: MetadataCache, config: KafkaConfig, configRepo
if (metadataCache.contains(topic)) {
val topicProps = configRepository.topicConfig(topic)
val logConfig = LogConfig.fromProps(config.extractLogConfigMap, topicProps)
- createResponseConfig(resource, logConfig, createTopicConfigEntry(logConfig, topicProps, includeSynonyms, includeDocumentation)(_, _))
+ // Internal configs are reported only when set on the topic itself, consistent with the CreateTopics response.
+ createResponseConfig(resource, logConfig, logConfig.overriddenConfigs, createTopicConfigEntry(logConfig, topicProps, includeSynonyms, includeDocumentation)(_, _))
} else {
new DescribeConfigsResponseData.DescribeConfigsResult().setErrorCode(Errors.UNKNOWN_TOPIC_OR_PARTITION.code)
.setConfigs(Collections.emptyList[DescribeConfigsResponseData.DescribeConfigsResourceResult])
diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala
index e156b28025e..ce4e67151ab 100755
--- a/core/src/main/scala/kafka/server/KafkaConfig.scala
+++ b/core/src/main/scala/kafka/server/KafkaConfig.scala
@@ -204,6 +204,8 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _], enforceProv
def messageMaxBytes = getInt(ServerConfigs.MESSAGE_MAX_BYTES_CONFIG)
+ def maxDecompressedMessageBytes = getInt(ServerConfigs.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG)
+
def getNumReplicaAlterLogDirsThreads: Int = {
val numThreads: Integer = Option(getInt(ServerConfigs.NUM_REPLICA_ALTER_LOG_DIRS_THREADS_CONFIG)).getOrElse(logDirs.size)
numThreads
@@ -622,6 +624,7 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _], enforceProv
logProps.put(TopicConfig.RETENTION_BYTES_CONFIG, logRetentionBytes)
logProps.put(TopicConfig.RETENTION_MS_CONFIG, logRetentionTimeMillis: java.lang.Long)
logProps.put(TopicConfig.MAX_MESSAGE_BYTES_CONFIG, messageMaxBytes)
+ logProps.put(TopicConfig.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG, maxDecompressedMessageBytes)
logProps.put(TopicConfig.INDEX_INTERVAL_BYTES_CONFIG, logIndexIntervalBytes)
logProps.put(TopicConfig.DELETE_RETENTION_MS_CONFIG, logCleanerDeleteRetentionMs)
logProps.put(TopicConfig.MIN_COMPACTION_LAG_MS_CONFIG, logCleanerMinCompactionLagMs)
diff --git a/core/src/test/scala/unit/kafka/log/LogConfigTest.scala b/core/src/test/scala/unit/kafka/log/LogConfigTest.scala
index 0086d4d34a9..5657e57cb48 100644
--- a/core/src/test/scala/unit/kafka/log/LogConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogConfigTest.scala
@@ -23,6 +23,7 @@ import org.apache.kafka.common.config.ConfigDef.Importance.MEDIUM
import org.apache.kafka.common.config.ConfigDef.Type.INT
import org.apache.kafka.common.config.{ConfigException, SslConfigs, TopicConfig}
import org.apache.kafka.common.errors.InvalidConfigurationException
+import org.apache.kafka.common.record.internal.Records
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.Test
@@ -82,6 +83,26 @@ class LogConfigTest {
})
}
+ @Test
+ def testMaxDecompressedMessageBytesProps(): Unit = {
+ // Default is the unlimited sentinel, so the limit is a no-op until an operator opts in.
+ assertEquals(Records.SOFT_MAX_ARRAY_LENGTH, LogConfig.DEFAULT_MAX_DECOMPRESSED_MESSAGE_BYTES)
+ assertEquals(Records.SOFT_MAX_ARRAY_LENGTH,
+ new LogConfig(new Properties()).maxDecompressedMessageBytes())
+
+ val props = new Properties()
+ props.put(TopicConfig.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG, "1000")
+ assertEquals(1000, new LogConfig(props).maxDecompressedMessageBytes())
+
+ // Values outside [1, SOFT_MAX_ARRAY_LENGTH] are rejected; the upper bound guarantees a validated
+ // config can never weaken the pre-allocation array-length guard in the record decoders.
+ for (invalid <- Seq("0", Int.MaxValue.toString)) {
+ val invalidProps = new Properties()
+ invalidProps.put(TopicConfig.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG, invalid)
+ assertThrows(classOf[ConfigException], () => new LogConfig(invalidProps))
+ }
+ }
+
@Test
def testInvalidCompactionLagConfig(): Unit = {
val props = new util.HashMap[String, String]
diff --git a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
index db7dbd8cab6..f08c53b4aa2 100644
--- a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
@@ -30,7 +30,7 @@ import org.apache.kafka.clients.consumer.AcknowledgeType
import org.apache.kafka.common._
import org.apache.kafka.common.acl.AclOperation
import org.apache.kafka.common.compress.Compression
-import org.apache.kafka.common.config.ConfigResource
+import org.apache.kafka.common.config.{ConfigResource, TopicConfig}
import org.apache.kafka.common.config.ConfigResource.Type.{BROKER, BROKER_LOGGER}
import org.apache.kafka.common.errors.{ClusterAuthorizationException, UnsupportedVersionException}
import org.apache.kafka.common.internals.{Plugin, Topic}
@@ -311,6 +311,38 @@ class KafkaApisTest extends Logging {
assertEquals(propValue, describeConfigsResponseData.value)
}
+ @Test
+ def testDescribeConfigsTopicIncludesInternalConfigSetOnTopic(): Unit = {
+ val resourceName = "topic-1"
+ val configRepository: ConfigRepository = mock(classOf[ConfigRepository])
+ val topicConfigs = new Properties()
+ topicConfigs.put(LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG, "1048576")
+ when(configRepository.topicConfig(resourceName)).thenReturn(topicConfigs)
+
+ metadataCache = mock(classOf[KRaftMetadataCache])
+ when(metadataCache.contains(resourceName)).thenReturn(true)
+
+ val requestHeader = new RequestHeader(ApiKeys.DESCRIBE_CONFIGS, ApiKeys.DESCRIBE_CONFIGS.latestVersion, clientId, 0)
+ val describeConfigsRequest = new DescribeConfigsRequest.Builder(new DescribeConfigsRequestData()
+ .setResources(util.List.of(new DescribeConfigsRequestData.DescribeConfigsResource()
+ .setResourceName(resourceName)
+ .setResourceType(ConfigResource.Type.TOPIC.id))))
+ .build(requestHeader.apiVersion)
+ val request = buildRequest(describeConfigsRequest, requestHeader = Option(requestHeader))
+
+ kafkaApis = createKafkaApis(configRepository = configRepository)
+ kafkaApis.handleDescribeConfigsRequest(request)
+
+ val response = verifyNoThrottling[DescribeConfigsResponse](request)
+ val configNames = response.data.results.get(0).configs.asScala.map(_.name).toSet
+ // Non-internal configs are always reported.
+ assertTrue(configNames.contains(TopicConfig.SEGMENT_BYTES_CONFIG))
+ // An internal config set on the topic itself is reported.
+ assertTrue(configNames.contains(LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG))
+ // A public config inherited from the broker-level default is reported as well.
+ assertTrue(configNames.contains(TopicConfig.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG))
+ }
+
@Test
def testElectLeadersForwarding(): Unit = {
val requestBuilder = new ElectLeadersRequest.Builder(ElectionType.PREFERRED, null, 30000)
diff --git a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
index 03af52c38d3..1718e728d07 100755
--- a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
@@ -1227,6 +1227,8 @@ class KafkaConfigTest {
assertDynamic(kafkaConfigProp, 10007, () => config.logIndexIntervalBytes)
case TopicConfig.MAX_MESSAGE_BYTES_CONFIG =>
assertDynamic(kafkaConfigProp, 10008, () => config.messageMaxBytes)
+ case TopicConfig.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG =>
+ assertDynamic(kafkaConfigProp, 20000, () => config.maxDecompressedMessageBytes)
case TopicConfig.MESSAGE_TIMESTAMP_BEFORE_MAX_MS_CONFIG =>
assertDynamic(kafkaConfigProp, 10015L, () => config.logMessageTimestampBeforeMaxMs)
case TopicConfig.MESSAGE_TIMESTAMP_AFTER_MAX_MS_CONFIG =>
diff --git a/server-common/src/main/java/org/apache/kafka/server/config/ServerConfigs.java b/server-common/src/main/java/org/apache/kafka/server/config/ServerConfigs.java
index 939feaf7248..76ea341091f 100644
--- a/server-common/src/main/java/org/apache/kafka/server/config/ServerConfigs.java
+++ b/server-common/src/main/java/org/apache/kafka/server/config/ServerConfigs.java
@@ -20,6 +20,7 @@ import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.common.config.TopicConfig;
import org.apache.kafka.common.record.internal.CompressionType;
+import org.apache.kafka.common.record.internal.Records;
import org.apache.kafka.server.authorizer.Authorizer;
import org.apache.kafka.server.record.BrokerCompressionType;
@@ -31,6 +32,7 @@ import static org.apache.kafka.common.config.ConfigDef.Importance.HIGH;
import static org.apache.kafka.common.config.ConfigDef.Importance.LOW;
import static org.apache.kafka.common.config.ConfigDef.Importance.MEDIUM;
import static org.apache.kafka.common.config.ConfigDef.Range.atLeast;
+import static org.apache.kafka.common.config.ConfigDef.Range.between;
import static org.apache.kafka.common.config.ConfigDef.Type.BOOLEAN;
import static org.apache.kafka.common.config.ConfigDef.Type.INT;
import static org.apache.kafka.common.config.ConfigDef.Type.LIST;
@@ -50,6 +52,9 @@ public class ServerConfigs {
public static final String MESSAGE_MAX_BYTES_CONFIG = "message.max.bytes";
public static final String MESSAGE_MAX_BYTES_DOC = TopicConfig.MAX_MESSAGE_BYTES_DOC +
"This can be set per topic with the topic level <code>" + TopicConfig.MAX_MESSAGE_BYTES_CONFIG + "</code> config.";
+ public static final String MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG = ServerTopicConfigSynonyms.serverSynonym(TopicConfig.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG);
+ public static final String MAX_DECOMPRESSED_MESSAGE_BYTES_DOC = TopicConfig.MAX_DECOMPRESSED_MESSAGE_BYTES_DOC +
+ " This can be set per topic with the topic level <code>" + TopicConfig.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG + "</code> config.";
public static final String NUM_IO_THREADS_CONFIG = "num.io.threads";
public static final int NUM_IO_THREADS_DEFAULT = 8;
@@ -132,6 +137,7 @@ public class ServerConfigs {
public static final ConfigDef CONFIG_DEF = new ConfigDef()
.define(BROKER_ID_CONFIG, INT, BROKER_ID_DEFAULT, HIGH, BROKER_ID_DOC)
.define(MESSAGE_MAX_BYTES_CONFIG, INT, ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT, atLeast(0), HIGH, MESSAGE_MAX_BYTES_DOC)
+ .define(MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG, INT, ServerLogConfigs.MAX_DECOMPRESSED_MESSAGE_BYTES_DEFAULT, between(1, Records.SOFT_MAX_ARRAY_LENGTH), MEDIUM, MAX_DECOMPRESSED_MESSAGE_BYTES_DOC)
.define(NUM_IO_THREADS_CONFIG, INT, NUM_IO_THREADS_DEFAULT, atLeast(1), HIGH, NUM_IO_THREADS_DOC)
.define(NUM_REPLICA_ALTER_LOG_DIRS_THREADS_CONFIG, INT, null, HIGH, NUM_REPLICA_ALTER_LOG_DIRS_THREADS_DOC)
.define(BACKGROUND_THREADS_CONFIG, INT, BACKGROUND_THREADS_DEFAULT, atLeast(1), HIGH, BACKGROUND_THREADS_DOC)
diff --git a/server-common/src/main/java/org/apache/kafka/server/config/ServerLogConfigs.java b/server-common/src/main/java/org/apache/kafka/server/config/ServerLogConfigs.java
index d8ffd8a5e2f..05430ce28cf 100644
--- a/server-common/src/main/java/org/apache/kafka/server/config/ServerLogConfigs.java
+++ b/server-common/src/main/java/org/apache/kafka/server/config/ServerLogConfigs.java
@@ -175,5 +175,6 @@ public class ServerLogConfigs {
"directory has failed for longer than this time, the broker will fail and shut down.";
public static final int MAX_MESSAGE_BYTES_DEFAULT = 1024 * 1024 + Records.LOG_OVERHEAD;
+ public static final int MAX_DECOMPRESSED_MESSAGE_BYTES_DEFAULT = Records.SOFT_MAX_ARRAY_LENGTH;
public static final String COMPRESSION_TYPE_DEFAULT = BrokerCompressionType.PRODUCER.name;
}
diff --git a/server-common/src/main/java/org/apache/kafka/server/config/ServerTopicConfigSynonyms.java b/server-common/src/main/java/org/apache/kafka/server/config/ServerTopicConfigSynonyms.java
index e195e7626de..9d02f014191 100644
--- a/server-common/src/main/java/org/apache/kafka/server/config/ServerTopicConfigSynonyms.java
+++ b/server-common/src/main/java/org/apache/kafka/server/config/ServerTopicConfigSynonyms.java
@@ -66,6 +66,7 @@ public final class ServerTopicConfigSynonyms {
new ConfigSynonym("retention.minutes", ConfigSynonym.MINUTES_TO_MILLISECONDS),
new ConfigSynonym("retention.hours", ConfigSynonym.HOURS_TO_MILLISECONDS)),
single(TopicConfig.MAX_MESSAGE_BYTES_CONFIG, "message.max.bytes"),
+ sameName(TopicConfig.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG),
sameNameWithLogPrefix(TopicConfig.INDEX_INTERVAL_BYTES_CONFIG),
sameNameWithLogCleanerPrefix(TopicConfig.DELETE_RETENTION_MS_CONFIG),
sameNameWithLogCleanerPrefix(TopicConfig.MIN_COMPACTION_LAG_MS_CONFIG),
diff --git a/server/src/main/java/org/apache/kafka/server/ConfigHelperUtils.java b/server/src/main/java/org/apache/kafka/server/ConfigHelperUtils.java
index b104de4c156..9ab6acb5b7b 100644
--- a/server/src/main/java/org/apache/kafka/server/ConfigHelperUtils.java
+++ b/server/src/main/java/org/apache/kafka/server/ConfigHelperUtils.java
@@ -24,6 +24,7 @@ import org.apache.kafka.common.protocol.Errors;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
+import java.util.Set;
import java.util.function.BiFunction;
import java.util.stream.Stream;
@@ -46,20 +47,36 @@ public class ConfigHelperUtils {
}
/**
- * Creates a DescribeConfigsResult from an AbstractConfig.
- * This method merges the config's originals (excluding nulls and keys present in nonInternalValues, which take priority).
+ * Creates a DescribeConfigsResult from an AbstractConfig, treating all of its originals as explicitly set.
*/
public static DescribeConfigsResponseData.DescribeConfigsResult createResponseConfig(
DescribeConfigsRequestData.DescribeConfigsResource resource,
AbstractConfig config,
BiFunction<String, Object, DescribeConfigsResponseData.DescribeConfigsResourceResult> createConfigEntry) {
+ return createResponseConfig(resource, config, config.originals().keySet(), createConfigEntry);
+ }
+
+ /**
+ * Creates a DescribeConfigsResult from an AbstractConfig whose originals may contain defaults merged with
+ * explicitly set configs, e.g. a topic config built from the broker's log defaults and the topic's own overrides.
+ * Non-internal configs are always included. Other originals, i.e. internal configs and any keys the
+ * ConfigDef does not define, are included only if explicitly set. This matches the CreateTopics response,
+ * which reports internal configs only when the user set them.
+ */
+ public static DescribeConfigsResponseData.DescribeConfigsResult createResponseConfig(
+ DescribeConfigsRequestData.DescribeConfigsResource resource,
+ AbstractConfig config,
+ Set<String> explicitlySetConfigs,
+ BiFunction<String, Object, DescribeConfigsResponseData.DescribeConfigsResourceResult> createConfigEntry) {
// Cast from Map<String, ?> to Map<String, Object> to eliminate wildcard types. Cached to avoid multiple calls.
@SuppressWarnings("unchecked")
Map<String, Object> nonInternalValues = (Map<String, Object>) config.nonInternalValues();
Stream<Entry<String, Object>> allEntries = Stream.concat(
config.originals().entrySet().stream()
- .filter(entry -> entry.getValue() != null && !nonInternalValues.containsKey(entry.getKey()))
+ .filter(entry -> entry.getValue() != null
+ && explicitlySetConfigs.contains(entry.getKey())
+ && !nonInternalValues.containsKey(entry.getKey()))
.map(entry -> Map.entry(entry.getKey(), entry.getValue())),
nonInternalValues.entrySet().stream()
);
diff --git a/server/src/test/java/org/apache/kafka/server/requests/ProduceRequestTest.java b/server/src/test/java/org/apache/kafka/server/requests/ProduceRequestTest.java
index 4dd43b609fc..e82165239b6 100644
--- a/server/src/test/java/org/apache/kafka/server/requests/ProduceRequestTest.java
+++ b/server/src/test/java/org/apache/kafka/server/requests/ProduceRequestTest.java
@@ -19,13 +19,17 @@ package org.apache.kafka.server.requests;
import kafka.server.KafkaBroker;
import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AlterConfigOp;
+import org.apache.kafka.clients.admin.ConfigEntry;
import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.TopicPartitionInfo;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.compress.Compression;
+import org.apache.kafka.common.config.ConfigResource;
import org.apache.kafka.common.config.TopicConfig;
import org.apache.kafka.common.message.ProduceRequestData;
+import org.apache.kafka.common.message.ProduceResponseData;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.record.TimestampType;
@@ -36,10 +40,12 @@ import org.apache.kafka.common.record.internal.SimpleRecord;
import org.apache.kafka.common.requests.ProduceRequest;
import org.apache.kafka.common.requests.ProduceResponse;
import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterConfigProperty;
import org.apache.kafka.common.test.api.ClusterTest;
import org.apache.kafka.common.test.api.ClusterTestDefaults;
import org.apache.kafka.server.IntegrationTestUtils;
import org.apache.kafka.server.metrics.KafkaYammerMetrics;
+import org.apache.kafka.test.TestUtils;
import com.yammer.metrics.core.Meter;
@@ -224,6 +230,153 @@ public class ProduceRequestTest {
assertEquals(-1L, partitionResponse.logAppendTimeMs());
}
+ // String so it can double as an @ClusterConfigProperty value (must be a compile-time constant).
+ private static final String SMALL_MAX_DECOMPRESSED_MESSAGE_BYTES = "512";
+ // Above the limit but tiny gzip-compressed and below max.message.bytes, so the decompressed
+ // per-record limit -- not the wire bound -- is what rejects it.
+ private static final int OVERSIZED_VALUE_BYTES = 4096;
+ private static final int UNDERSIZED_VALUE_BYTES = 64;
+
+ /**
+ * A topic-level limit rejects a compressed record whose declared decompressed body exceeds it
+ * (INVALID_RECORD, before allocating the body). A small compressed record and an equally-large
+ * uncompressed record (bounded instead by max.message.bytes) are accepted.
+ */
+ @ClusterTest
+ public void testProduceRejectsCompressedRecordExceedingMaxDecompressedMessageBytes() throws Exception {
+ cluster.createTopic(TOPIC, 1, (short) 1,
+ Map.of(TopicConfig.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG, SMALL_MAX_DECOMPRESSED_MESSAGE_BYTES));
+ int leaderId = cluster.getLeaderBrokerId(new TopicPartition(TOPIC, 0));
+ Uuid topicId = getTopicId();
+
+ var rejected = onlyPartitionResponse(sendProduceRequest(leaderId,
+ produceRequest(topicId, singleRecord(Compression.gzip().build(), OVERSIZED_VALUE_BYTES))));
+ assertEquals(Errors.INVALID_RECORD.code(), rejected.errorCode(),
+ "a compressed record exceeding the configured per-record limit must be rejected as invalid");
+ assertEquals(-1L, rejected.baseOffset(), "a rejected record must not be appended");
+
+ var acceptedSmall = onlyPartitionResponse(sendProduceRequest(leaderId,
+ produceRequest(topicId, singleRecord(Compression.gzip().build(), UNDERSIZED_VALUE_BYTES))));
+ assertEquals(Errors.NONE.code(), acceptedSmall.errorCode(),
+ "a compressed record under the configured per-record limit must be accepted");
+
+ var acceptedUncompressed = onlyPartitionResponse(sendProduceRequest(leaderId,
+ produceRequest(topicId, singleRecord(Compression.NONE, OVERSIZED_VALUE_BYTES))));
+ assertEquals(Errors.NONE.code(), acceptedUncompressed.errorCode(),
+ "an uncompressed record must not be subject to the decompressed per-record limit");
+ }
+
+ /**
+ * The topic-level limit is dynamically reconfigurable: a large compressed record is accepted at
+ * the default, then rejected after lowering it via incrementalAlterConfigs -- no restart.
+ */
+ @ClusterTest
+ public void testMaxDecompressedMessageBytesIsDynamicallyReconfigurable() throws Exception {
+ cluster.createTopic(TOPIC, 1, (short) 1);
+ int leaderId = cluster.getLeaderBrokerId(new TopicPartition(TOPIC, 0));
+ Uuid topicId = getTopicId();
+
+ var before = onlyPartitionResponse(sendProduceRequest(leaderId,
+ produceRequest(topicId, singleRecord(Compression.gzip().build(), OVERSIZED_VALUE_BYTES))));
+ assertEquals(Errors.NONE.code(), before.errorCode(),
+ "with the default per-record limit the record must be accepted");
+
+ try (Admin admin = cluster.admin()) {
+ ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC, TOPIC);
+ admin.incrementalAlterConfigs(Map.of(resource, List.of(new AlterConfigOp(
+ new ConfigEntry(TopicConfig.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG, SMALL_MAX_DECOMPRESSED_MESSAGE_BYTES),
+ AlterConfigOp.OpType.SET)))).all().get();
+ }
+
+ // The topic-config change reaches the produce path asynchronously; poll until it takes effect.
+ TestUtils.waitForCondition(
+ () -> onlyPartitionResponse(sendProduceRequest(leaderId,
+ produceRequest(topicId, singleRecord(Compression.gzip().build(), OVERSIZED_VALUE_BYTES))))
+ .errorCode() == Errors.INVALID_RECORD.code(),
+ 15000L,
+ "the lowered topic-level " + TopicConfig.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG
+ + " was not applied to the produce path");
+ }
+
+ /**
+ * A broker-level default is inherited by a topic without an override: an oversized compressed
+ * record is rejected.
+ */
+ @ClusterTest(serverProperties = {
+ @ClusterConfigProperty(key = TopicConfig.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG, value = SMALL_MAX_DECOMPRESSED_MESSAGE_BYTES)
+ })
+ public void testBrokerDefaultMaxDecompressedMessageBytesAppliesToTopicWithoutOverride() throws Exception {
+ cluster.createTopic(TOPIC, 1, (short) 1);
+ int leaderId = cluster.getLeaderBrokerId(new TopicPartition(TOPIC, 0));
+ Uuid topicId = getTopicId();
+
+ var rejected = onlyPartitionResponse(sendProduceRequest(leaderId,
+ produceRequest(topicId, singleRecord(Compression.gzip().build(), OVERSIZED_VALUE_BYTES))));
+ assertEquals(Errors.INVALID_RECORD.code(), rejected.errorCode(),
+ "the broker-level default per-record limit must apply to a topic without an override");
+ }
+
+ /**
+ * The broker-level default is dynamically reconfigurable and flows to a topic without an
+ * override: accepted at the default, rejected after lowering the cluster-wide default via a
+ * BROKER-resource alter (asserted dynamic in KafkaConfigTest#testDynamicLogConfigs) -- no restart.
+ */
+ @ClusterTest
+ public void testBrokerDefaultMaxDecompressedMessageBytesIsDynamicallyReconfigurable() throws Exception {
+ cluster.createTopic(TOPIC, 1, (short) 1);
+ int leaderId = cluster.getLeaderBrokerId(new TopicPartition(TOPIC, 0));
+ Uuid topicId = getTopicId();
+
+ var before = onlyPartitionResponse(sendProduceRequest(leaderId,
+ produceRequest(topicId, singleRecord(Compression.gzip().build(), OVERSIZED_VALUE_BYTES))));
+ assertEquals(Errors.NONE.code(), before.errorCode(),
+ "with the default broker per-record limit and no topic override the record must be accepted");
+
+ try (Admin admin = cluster.admin()) {
+ // Empty resource name = cluster-wide broker default.
+ ConfigResource resource = new ConfigResource(ConfigResource.Type.BROKER, "");
+ admin.incrementalAlterConfigs(Map.of(resource, List.of(new AlterConfigOp(
+ new ConfigEntry(TopicConfig.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG, SMALL_MAX_DECOMPRESSED_MESSAGE_BYTES),
+ AlterConfigOp.OpType.SET)))).all().get();
+ }
+
+ // The broker-default change reaches the produce path asynchronously; poll until it takes effect.
+ TestUtils.waitForCondition(
+ () -> onlyPartitionResponse(sendProduceRequest(leaderId,
+ produceRequest(topicId, singleRecord(Compression.gzip().build(), OVERSIZED_VALUE_BYTES))))
+ .errorCode() == Errors.INVALID_RECORD.code(),
+ 15000L,
+ "the lowered cluster-wide broker-default " + TopicConfig.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG
+ + " was not applied to the produce path");
+ }
+
+ private ProduceRequest produceRequest(Uuid topicId, MemoryRecords records) {
+ return ProduceRequest.builder(new ProduceRequestData()
+ .setTopicData(new ProduceRequestData.TopicProduceDataCollection(Collections.singletonList(
+ new ProduceRequestData.TopicProduceData()
+ .setTopicId(topicId)
+ .setPartitionData(Collections.singletonList(
+ new ProduceRequestData.PartitionProduceData()
+ .setIndex(0)
+ .setRecords(records))))
+ .iterator()))
+ .setAcks((short) -1)
+ .setTimeoutMs(3000)
+ .setTransactionalId(null)).build();
+ }
+
+ private static MemoryRecords singleRecord(Compression compression, int valueSize) {
+ return MemoryRecords.withRecords(compression,
+ new SimpleRecord(System.currentTimeMillis(), "key".getBytes(), new byte[valueSize]));
+ }
+
+ private ProduceResponseData.PartitionProduceResponse onlyPartitionResponse(ProduceResponse response) {
+ assertEquals(1, response.data().responses().size());
+ var topicResponse = response.data().responses().iterator().next();
+ assertEquals(1, topicResponse.partitionResponses().size());
+ return topicResponse.partitionResponses().get(0);
+ }
+
private ProduceResponse sendProduceRequest(int brokerId, ProduceRequest request) throws IOException {
KafkaBroker broker = cluster.brokers().get(brokerId);
int port = broker.socketServer().boundPort(cluster.clientListener());
diff --git a/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java b/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java
index 1ac6028af51..ac20d02cf03 100644
--- a/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java
+++ b/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java
@@ -643,8 +643,8 @@ public class RemoteLogManager implements Closeable, AsyncOffsetReader {
return remoteLogMetadataManagerPlugin.get().nextSegmentWithTxnIndex(tpId, epochForOffset, offset);
}
- Optional<FileRecords.TimestampAndOffset> lookupTimestamp(RemoteLogSegmentMetadata rlsMetadata, long timestamp, long startingOffset)
- throws RemoteStorageException, IOException {
+ Optional<FileRecords.TimestampAndOffset> lookupTimestamp(RemoteLogSegmentMetadata rlsMetadata, long timestamp, long startingOffset,
+ int maxRecordBodySize) throws RemoteStorageException, IOException {
int startPos = indexCache.lookupTimestamp(rlsMetadata, timestamp, startingOffset);
InputStream remoteSegInputStream = null;
@@ -657,7 +657,7 @@ public class RemoteLogManager implements Closeable, AsyncOffsetReader {
RecordBatch batch = remoteLogInputStream.nextBatch();
if (batch == null) break;
if (batch.maxTimestamp() >= timestamp && batch.lastOffset() >= startingOffset) {
- try (CloseableIterator<Record> recordStreamingIterator = batch.streamingIterator(BufferSupplier.NO_CACHING)) {
+ try (CloseableIterator<Record> recordStreamingIterator = batch.streamingIterator(BufferSupplier.NO_CACHING, maxRecordBodySize)) {
while (recordStreamingIterator.hasNext()) {
Record record = recordStreamingIterator.next();
if (record.timestamp() >= timestamp && record.offset() >= startingOffset)
@@ -731,6 +731,7 @@ public class RemoteLogManager implements Closeable, AsyncOffsetReader {
throw new KafkaException("UnifiedLog does not exist for topic partition: " + tp);
}
UnifiedLog unifiedLog = unifiedLogOptional.get();
+ int maxRecordBodySize = unifiedLog.config().maxDecompressedMessageBytes();
// Get the respective epoch in which the starting-offset exists.
OptionalInt maybeEpoch = leaderEpochCache.epochForOffset(startingOffset);
@@ -751,12 +752,12 @@ public class RemoteLogManager implements Closeable, AsyncOffsetReader {
List<LogSegment> segmentsCopy = unifiedLog.logSegments();
if (segmentsCopy.isEmpty() || rlsMetadata.startOffset() < segmentsCopy.get(0).baseOffset()) {
// search in remote-log
- return lookupTimestamp(rlsMetadata, timestamp, startingOffset);
+ return lookupTimestamp(rlsMetadata, timestamp, startingOffset, maxRecordBodySize);
} else {
// search in local-log
for (LogSegment segment : segmentsCopy) {
if (segment.largestTimestamp() >= timestamp) {
- return segment.findOffsetByTimestamp(timestamp, startingOffset);
+ return segment.findOffsetByTimestamp(timestamp, startingOffset, maxRecordBodySize);
}
}
}
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java
index a2c0399bd70..72305608efa 100644
--- a/storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/Cleaner.java
@@ -254,7 +254,8 @@ public class Cleaner {
lastOffsetOfActiveProducers,
log.highWatermark(),
stats,
- currentTime
+ currentTime,
+ log.config().maxDecompressedMessageBytes()
);
} catch (LogSegmentOffsetOverflowException e) {
// Split the current segment. It's also safest to abort the current cleaning process, so that we retry from
@@ -305,6 +306,8 @@ public class Cleaner {
* the high watermark so that the last offset information is not lost after cleaning
* @param stats Collector for cleaning statistics
* @param currentTime The time at which the clean was initiated
+ * @param maxRecordBodySize The maximum decompressed per-record body size of the corresponding topic; records
+ * exceeding it are rejected with an InvalidRecordException before allocation
*/
private void cleanInto(TopicPartition topicPartition,
FileRecords sourceRecords,
@@ -317,7 +320,8 @@ public class Cleaner {
Map<Long, LastRecord> lastRecordsOfActiveProducers,
long highWatermark,
CleanerStats stats,
- long currentTime) throws IOException {
+ long currentTime,
+ int maxRecordBodySize) throws IOException {
MemoryRecords.RecordFilter logCleanerFilter = new MemoryRecords.RecordFilter(currentTime, deleteRetentionMs) {
private boolean discardBatchRecords;
@@ -391,7 +395,7 @@ public class Cleaner {
sourceRecords.readInto(readBuffer, position);
MemoryRecords records = MemoryRecords.readableRecords(readBuffer);
throttler.maybeThrottle(records.sizeInBytes());
- MemoryRecords.FilterResult result = records.filterTo(logCleanerFilter, writeBuffer, decompressionBufferSupplier);
+ MemoryRecords.FilterResult result = records.filterTo(logCleanerFilter, writeBuffer, decompressionBufferSupplier, maxRecordBodySize);
stats.readMessages(result.messagesRead(), result.bytesRead());
stats.recopyMessages(result.messagesRetained(), result.bytesRetained());
@@ -672,7 +676,8 @@ public class Cleaner {
nextSegmentStartOffset,
log.config().maxMessageSize(),
transactionMetadata,
- stats
+ stats,
+ log.config().maxDecompressedMessageBytes()
);
if (full) {
logger.debug("Offset map is full, {} segments fully mapped, segment with base offset {} is partially mapped",
@@ -694,6 +699,8 @@ public class Cleaner {
* @param maxLogMessageSize The maximum size in bytes for record allowed
* @param transactionMetadata The state of ongoing transactions for the log between offset range to build
* @param stats Collector for cleaning statistics
+ * @param maxRecordBodySize The maximum decompressed per-record body size of the corresponding topic; records
+ * exceeding it are rejected with an InvalidRecordException before allocation
*
* @return If the map was filled whilst loading from this segment
*/
@@ -704,7 +711,8 @@ public class Cleaner {
long nextSegmentStartOffset,
int maxLogMessageSize,
CleanedTransactionMetadata transactionMetadata,
- CleanerStats stats) throws IOException, DigestException {
+ CleanerStats stats,
+ int maxRecordBodySize) throws IOException, DigestException {
int position = segment.offsetIndex().lookup(startOffset).position();
int maxDesiredMapSize = (int) (map.slots() * dupBufferLoadFactor);
@@ -732,7 +740,7 @@ public class Cleaner {
// Note that abort markers are supported in v2 and above, which means count is defined.
stats.indexMessagesRead(batch.countOrNull());
} else {
- try (CloseableIterator<Record> recordsIterator = batch.streamingIterator(decompressionBufferSupplier)) {
+ try (CloseableIterator<Record> recordsIterator = batch.streamingIterator(decompressionBufferSupplier, maxRecordBodySize)) {
for (Record record : (Iterable<Record>) () -> recordsIterator) {
if (record.hasKey() && record.offset() >= startOffset) {
if (map.size() < maxDesiredMapSize) {
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java
index f4a1865f26d..cbce662af9c 100644
--- a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java
@@ -27,6 +27,7 @@ import org.apache.kafka.common.config.TopicConfig;
import org.apache.kafka.common.errors.InvalidConfigurationException;
import org.apache.kafka.common.record.TimestampType;
import org.apache.kafka.common.record.internal.CompressionType;
+import org.apache.kafka.common.record.internal.Records;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.common.utils.internals.ConfigUtils;
import org.apache.kafka.server.config.QuotaConfig;
@@ -143,6 +144,10 @@ public class LogConfig extends AbstractConfig {
public static final boolean DEFAULT_REMOTE_LOG_DELETE_ON_DISABLE = false;
public static final long DEFAULT_LOCAL_RETENTION_BYTES = -2; // It indicates the value to be derived from RetentionBytes
public static final long DEFAULT_LOCAL_RETENTION_MS = -2; // It indicates the value to be derived from RetentionMs
+ // Default is the JVM array-length limit, i.e. effectively unlimited / no additional limit beyond
+ // the existing array-length OutOfMemoryError guard, so the per-record limit only takes effect when
+ // an operator configures it below this value.
+ public static final int DEFAULT_MAX_DECOMPRESSED_MESSAGE_BYTES = ServerLogConfigs.MAX_DECOMPRESSED_MESSAGE_BYTES_DEFAULT;
public static final String INTERNAL_SEGMENT_BYTES_CONFIG = "internal.segment.bytes";
public static final String INTERNAL_SEGMENT_BYTES_DOC = "The maximum size of a single log file. This should be used for testing only.";
@@ -208,6 +213,8 @@ public class LogConfig extends AbstractConfig {
TopicConfig.RETENTION_MS_DOC)
.define(TopicConfig.MAX_MESSAGE_BYTES_CONFIG, INT, ServerLogConfigs.MAX_MESSAGE_BYTES_DEFAULT, atLeast(0), MEDIUM,
TopicConfig.MAX_MESSAGE_BYTES_DOC)
+ .define(TopicConfig.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG, INT, DEFAULT_MAX_DECOMPRESSED_MESSAGE_BYTES,
+ between(1, Records.SOFT_MAX_ARRAY_LENGTH), MEDIUM, TopicConfig.MAX_DECOMPRESSED_MESSAGE_BYTES_DOC)
.define(TopicConfig.INDEX_INTERVAL_BYTES_CONFIG, INT, ServerLogConfigs.LOG_INDEX_INTERVAL_BYTES_DEFAULT, atLeast(0), MEDIUM,
TopicConfig.INDEX_INTERVAL_BYTES_DOC)
.define(TopicConfig.DELETE_RETENTION_MS_CONFIG, LONG, DEFAULT_DELETE_RETENTION_MS, atLeast(0), MEDIUM,
@@ -298,6 +305,7 @@ public class LogConfig extends AbstractConfig {
private final RemoteLogConfig remoteLogConfig;
private final int maxMessageSize;
+ private final int maxDecompressedMessageBytes;
private final Map<?, ?> props;
public LogConfig(Map<?, ?> props) {
@@ -320,6 +328,7 @@ public class LogConfig extends AbstractConfig {
this.retentionSize = getLong(TopicConfig.RETENTION_BYTES_CONFIG);
this.retentionMs = getLong(TopicConfig.RETENTION_MS_CONFIG);
this.maxMessageSize = getInt(TopicConfig.MAX_MESSAGE_BYTES_CONFIG);
+ this.maxDecompressedMessageBytes = getInt(TopicConfig.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG);
this.indexInterval = getInt(TopicConfig.INDEX_INTERVAL_BYTES_CONFIG);
this.fileDeleteDelayMs = getLong(TopicConfig.FILE_DELETE_DELAY_MS_CONFIG);
this.deleteRetentionMs = getLong(TopicConfig.DELETE_RETENTION_MS_CONFIG);
@@ -376,6 +385,11 @@ public class LogConfig extends AbstractConfig {
return maxMessageSize;
}
+ // Exposed as a method so it can be mocked
+ public int maxDecompressedMessageBytes() {
+ return maxDecompressedMessageBytes;
+ }
+
public long randomSegmentJitter() {
if (segmentJitterMs == 0)
return 0;
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegment.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegment.java
index 0effe0ecd4d..7169a9f9f0e 100644
--- a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegment.java
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegment.java
@@ -750,15 +750,17 @@ public class LogSegment implements Closeable {
*
* @param timestampMs The timestamp to search for.
* @param startingOffset The starting offset to search.
+ * @param maxRecordBodySize The maximum declared (decompressed) body size of a single record; a compressed record
+ * exceeding it is rejected with an InvalidRecordException before its body is allocated.
* @return the timestamp and offset of the first message that meets the requirements. Empty will be returned if there is no such message.
*/
- public Optional<FileRecords.TimestampAndOffset> findOffsetByTimestamp(long timestampMs, long startingOffset) throws IOException {
+ public Optional<FileRecords.TimestampAndOffset> findOffsetByTimestamp(long timestampMs, long startingOffset, int maxRecordBodySize) throws IOException {
// Get the index entry with a timestamp less than or equal to the target timestamp
TimestampOffset timestampOffset = timeIndex().lookup(timestampMs);
int position = offsetIndex().lookup(Math.max(timestampOffset.offset(), startingOffset)).position();
// Search the timestamp
- return Optional.ofNullable(log.searchForTimestamp(timestampMs, position, startingOffset));
+ return Optional.ofNullable(log.searchForTimestamp(timestampMs, position, startingOffset, maxRecordBodySize));
}
/**
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogValidator.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogValidator.java
index 2f998ed283a..31ee8997548 100644
--- a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogValidator.java
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogValidator.java
@@ -32,6 +32,7 @@ import org.apache.kafka.common.record.internal.MemoryRecordsBuilder.RecordsInfo;
import org.apache.kafka.common.record.internal.MutableRecordBatch;
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.ProduceResponse.RecordError;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.internals.BufferSupplier;
@@ -78,6 +79,7 @@ public class LogValidator {
private final long timestampAfterMaxMs;
private final int partitionLeaderEpoch;
private final AppendOrigin origin;
+ private final int maxRecordBodySize;
public LogValidator(MemoryRecords records,
TopicPartition topicPartition,
@@ -91,6 +93,24 @@ public class LogValidator {
long timestampAfterMaxMs,
int partitionLeaderEpoch,
AppendOrigin origin) {
+ this(records, topicPartition, time, sourceCompressionType, targetCompression, compactedTopic, toMagic,
+ timestampType, timestampBeforeMaxMs, timestampAfterMaxMs, partitionLeaderEpoch, origin,
+ Records.SOFT_MAX_ARRAY_LENGTH);
+ }
+
+ public LogValidator(MemoryRecords records,
+ TopicPartition topicPartition,
+ Time time,
+ CompressionType sourceCompressionType,
+ Compression targetCompression,
+ boolean compactedTopic,
+ byte toMagic,
+ TimestampType timestampType,
+ long timestampBeforeMaxMs,
+ long timestampAfterMaxMs,
+ int partitionLeaderEpoch,
+ AppendOrigin origin,
+ int maxRecordBodySize) {
this.records = records;
this.topicPartition = topicPartition;
this.time = time;
@@ -103,6 +123,7 @@ public class LogValidator {
this.timestampAfterMaxMs = timestampAfterMaxMs;
this.partitionLeaderEpoch = partitionLeaderEpoch;
this.origin = origin;
+ this.maxRecordBodySize = maxRecordBodySize;
}
/**
@@ -312,9 +333,9 @@ public class LogValidator {
// then we can optimize the iterator to skip key / value / headers since they would not be used at all
CloseableIterator<Record> recordsIterator;
if (inPlaceAssignment && firstBatch.magic() >= RecordBatch.MAGIC_VALUE_V2)
- recordsIterator = batch.skipKeyValueIterator(bufferSupplier);
+ recordsIterator = batch.skipKeyValueIterator(bufferSupplier, maxRecordBodySize);
else
- recordsIterator = batch.streamingIterator(bufferSupplier);
+ recordsIterator = batch.streamingIterator(bufferSupplier, maxRecordBodySize);
try {
List<ApiRecordError> recordErrors = new ArrayList<>(0);
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java
index 0d4a2d65625..3f49fccfb23 100644
--- a/storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java
@@ -1156,7 +1156,8 @@ public class UnifiedLog implements AutoCloseable {
config().messageTimestampBeforeMaxMs,
config().messageTimestampAfterMaxMs,
leaderEpoch,
- origin
+ origin,
+ config().maxDecompressedMessageBytes()
);
LogValidator.ValidationResult validateAndOffsetAssignResult = validator.validateMessagesAndAssignOffsets(offset,
validatorMetricsRecorder,
@@ -1741,7 +1742,7 @@ public class UnifiedLog implements AutoCloseable {
Optional<FileRecords.TimestampAndOffset> timestampAndOffsetOpt = findFirst(
latestTimestampSegment.log().batchesFrom(position.position()),
item -> item.maxTimestamp() == maxTimestampSoFar.timestamp())
- .flatMap(batch -> batch.offsetOfMaxTimestamp()
+ .flatMap(batch -> batch.offsetOfMaxTimestamp(config().maxDecompressedMessageBytes())
.map(offset -> new FileRecords.TimestampAndOffset(
batch.maxTimestamp(),
offset,
@@ -1803,7 +1804,7 @@ public class UnifiedLog implements AutoCloseable {
List<LogSegment> segments = logSegments();
for (LogSegment segment : segments) {
if (segment.largestTimestamp() >= targetTimestamp) {
- return segment.findOffsetByTimestamp(targetTimestamp, startOffset);
+ return segment.findOffsetByTimestamp(targetTimestamp, startOffset, config().maxDecompressedMessageBytes());
}
}
return Optional.empty();
diff --git a/storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerTest.java b/storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerTest.java
index 9f1f15a8872..d681428bae6 100644
--- a/storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerTest.java
+++ b/storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerTest.java
@@ -17,12 +17,14 @@
package org.apache.kafka.server.log.remote.storage;
import org.apache.kafka.common.Endpoint;
+import org.apache.kafka.common.InvalidRecordException;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.TopicIdPartition;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.compress.Compression;
import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.TopicConfig;
import org.apache.kafka.common.errors.ReplicaNotAvailableException;
import org.apache.kafka.common.metrics.KafkaMetric;
import org.apache.kafka.common.metrics.Metrics;
@@ -1807,7 +1809,7 @@ public class RemoteLogManagerTest {
return remoteLogMetadataManager;
}
@Override
- Optional<FileRecords.TimestampAndOffset> lookupTimestamp(RemoteLogSegmentMetadata rlsMetadata, long timestamp, long startingOffset) {
+ Optional<FileRecords.TimestampAndOffset> lookupTimestamp(RemoteLogSegmentMetadata rlsMetadata, long timestamp, long startingOffset, int maxRecordBodySize) {
return Optional.of(expectedRemoteResult);
}
};
@@ -1837,7 +1839,7 @@ public class RemoteLogManagerTest {
when(logSegment.baseOffset()).thenReturn(baseOffset);
when(logSegment.largestTimestamp()).thenReturn(largestTimestamp);
if (timestampAndOffset != null) {
- when(logSegment.findOffsetByTimestamp(anyLong(), anyLong()))
+ when(logSegment.findOffsetByTimestamp(anyLong(), anyLong(), anyInt()))
.thenReturn(Optional.of(timestampAndOffset));
}
return logSegment;
@@ -4377,4 +4379,39 @@ public class RemoteLogManagerTest {
pluginMetrics = true;
}
}
+
+
+ @Test
+ void testFindOffsetByTimestampRejectsRemoteRecordExceedingMaxDecompressedMessageBytes() throws IOException, RemoteStorageException {
+ TopicPartition tp = leaderTopicIdPartition.topicPartition();
+ long ts = time.milliseconds();
+ long startOffset = 120;
+ int targetLeaderEpoch = 10;
+
+ TreeMap<Integer, Long> validSegmentEpochs = new TreeMap<>();
+ validSegmentEpochs.put(targetLeaderEpoch, startOffset);
+
+ LeaderEpochFileCache leaderEpochFileCache = new LeaderEpochFileCache(tp, checkpoint, scheduler);
+ leaderEpochFileCache.assign(4, 99L);
+ leaderEpochFileCache.assign(5, 99L);
+ leaderEpochFileCache.assign(targetLeaderEpoch, startOffset);
+ leaderEpochFileCache.assign(12, 500L);
+
+ doTestFindOffsetByTimestamp(ts, startOffset, targetLeaderEpoch, validSegmentEpochs, RemoteLogSegmentState.COPY_SEGMENT_FINISHED);
+
+ // Serve a remote segment holding a compressed record whose decompressed body exceeds the topic's limit
+ MemoryRecords oversized = MemoryRecords.withRecords(startOffset, Compression.gzip().build(), targetLeaderEpoch,
+ new SimpleRecord(ts + 1, "key".getBytes(), new byte[1000]));
+ byte[] oversizedBytes = new byte[oversized.sizeInBytes()];
+ oversized.buffer().get(oversizedBytes);
+ when(remoteStorageManager.fetchLogSegment(any(RemoteLogSegmentMetadata.class), anyInt()))
+ .thenAnswer(a -> new ByteArrayInputStream(oversizedBytes));
+ Properties props = new Properties();
+ props.put(TopicConfig.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG, "100");
+ when(mockLog.config()).thenReturn(new LogConfig(props));
+
+ InvalidRecordException e = assertThrows(InvalidRecordException.class,
+ () -> remoteLogManager.findOffsetByTimestamp(tp, ts, startOffset, leaderEpochFileCache));
+ assertTrue(e.getMessage().contains("exceeds the configured maximum record size of 100"), e.getMessage());
+ }
}
diff --git a/storage/src/test/java/org/apache/kafka/storage/internals/log/LogCleanerTest.java b/storage/src/test/java/org/apache/kafka/storage/internals/log/LogCleanerTest.java
index 7c1a086d516..8d15ccee4f7 100644
--- a/storage/src/test/java/org/apache/kafka/storage/internals/log/LogCleanerTest.java
+++ b/storage/src/test/java/org/apache/kafka/storage/internals/log/LogCleanerTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.storage.internals.log;
+import org.apache.kafka.common.InvalidRecordException;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.compress.Compression;
import org.apache.kafka.common.config.AbstractConfig;
@@ -355,6 +356,33 @@ public class LogCleanerTest {
"Cleaned segment file should be trimmed to its real size.");
}
+ /**
+ * Compaction rejects a compressed record whose decompressed body exceeds
+ * max.decompressed.message.bytes with InvalidRecordException (mapped to an uncleanable
+ * partition by LogCleaner) rather than OOMing. Injected via follower append to skip produce
+ * validation, mirroring a record that was already durable before the limit was lowered.
+ */
+ @Test
+ public void testCleanRejectsRecordExceedingConfiguredMaxDecompressedMessageBytes() throws IOException {
+ Cleaner cleaner = makeCleaner(10);
+ Properties logProps = new Properties();
+ logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT);
+ logProps.put(TopicConfig.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG, "10");
+ UnifiedLog log = makeLog(LogConfig.fromProps(logConfig.originals(), logProps));
+
+ // partitionLeaderEpoch must be valid (>= 0) for follower append to assign it to the epoch cache.
+ MemoryRecords oversized = MemoryRecords.withIdempotentRecords(RecordBatch.CURRENT_MAGIC_VALUE, 0L,
+ Compression.gzip().build(), RecordBatch.NO_PRODUCER_ID, RecordBatch.NO_PRODUCER_EPOCH,
+ RecordBatch.NO_SEQUENCE, 0, new SimpleRecord("k".getBytes(), new byte[100]));
+ log.appendAsFollower(oversized, Integer.MAX_VALUE);
+ log.roll();
+
+ InvalidRecordException e = assertThrows(InvalidRecordException.class,
+ () -> cleaner.clean(new LogToClean(log, 0L, log.activeSegment().baseOffset(), false)));
+ assertTrue(e.getMessage().contains("exceeds the configured maximum record size"),
+ "expected the configured-maximum guard message, got: " + e.getMessage());
+ }
+
@Test
public void testDuplicateCheckAfterCleaning() throws IOException, DigestException {
Cleaner cleaner = makeCleaner(Integer.MAX_VALUE);
diff --git a/storage/src/test/java/org/apache/kafka/storage/internals/log/LogSegmentTest.java b/storage/src/test/java/org/apache/kafka/storage/internals/log/LogSegmentTest.java
index 9af969fe82d..f2a8829f721 100644
--- a/storage/src/test/java/org/apache/kafka/storage/internals/log/LogSegmentTest.java
+++ b/storage/src/test/java/org/apache/kafka/storage/internals/log/LogSegmentTest.java
@@ -387,17 +387,17 @@ public class LogSegmentTest {
assertEquals(490, seg.largestTimestamp());
// Search for an indexed timestamp
- assertEquals(42, seg.findOffsetByTimestamp(420, 0L).get().offset);
- assertEquals(43, seg.findOffsetByTimestamp(421, 0L).get().offset);
+ assertEquals(42, seg.findOffsetByTimestamp(420, 0L, Records.SOFT_MAX_ARRAY_LENGTH).get().offset);
+ assertEquals(43, seg.findOffsetByTimestamp(421, 0L, Records.SOFT_MAX_ARRAY_LENGTH).get().offset);
// Search for an un-indexed timestamp
- assertEquals(43, seg.findOffsetByTimestamp(430, 0L).get().offset);
- assertEquals(44, seg.findOffsetByTimestamp(431, 0L).get().offset);
+ assertEquals(43, seg.findOffsetByTimestamp(430, 0L, Records.SOFT_MAX_ARRAY_LENGTH).get().offset);
+ assertEquals(44, seg.findOffsetByTimestamp(431, 0L, Records.SOFT_MAX_ARRAY_LENGTH).get().offset);
// Search beyond the last timestamp
- assertEquals(Optional.empty(), seg.findOffsetByTimestamp(491, 0L));
+ assertEquals(Optional.empty(), seg.findOffsetByTimestamp(491, 0L, Records.SOFT_MAX_ARRAY_LENGTH));
// Search before the first indexed timestamp
- assertEquals(41, seg.findOffsetByTimestamp(401, 0L).get().offset);
+ assertEquals(41, seg.findOffsetByTimestamp(401, 0L, Records.SOFT_MAX_ARRAY_LENGTH).get().offset);
// Search before the first timestamp
- assertEquals(40, seg.findOffsetByTimestamp(399, 0L).get().offset);
+ assertEquals(40, seg.findOffsetByTimestamp(399, 0L, Records.SOFT_MAX_ARRAY_LENGTH).get().offset);
}
}
@@ -590,9 +590,9 @@ public class LogSegmentTest {
writeNonsenseToFile(timeIndexFile, 5, (int) timeIndexFile.length());
seg.recover(newProducerStateManager(), mock(LeaderEpochFileCache.class));
for (int i = 0; i < 100; i++) {
- assertEquals(i, seg.findOffsetByTimestamp(i * 10, 0L).get().offset);
+ assertEquals(i, seg.findOffsetByTimestamp(i * 10, 0L, Records.SOFT_MAX_ARRAY_LENGTH).get().offset);
if (i < 99) {
- assertEquals(i + 1, seg.findOffsetByTimestamp(i * 10 + 1, 0L).get().offset);
+ assertEquals(i + 1, seg.findOffsetByTimestamp(i * 10 + 1, 0L, Records.SOFT_MAX_ARRAY_LENGTH).get().offset);
}
}
}
diff --git a/storage/src/test/java/org/apache/kafka/storage/internals/log/LogTestUtils.java b/storage/src/test/java/org/apache/kafka/storage/internals/log/LogTestUtils.java
index 6939d977f4c..e1fe3898f52 100644
--- a/storage/src/test/java/org/apache/kafka/storage/internals/log/LogTestUtils.java
+++ b/storage/src/test/java/org/apache/kafka/storage/internals/log/LogTestUtils.java
@@ -625,6 +625,11 @@ public class LogTestUtils {
return this;
}
+ public LogConfigBuilder maxDecompressedMessageBytes(int maxDecompressedMessageBytes) {
+ configs.put(TopicConfig.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG, maxDecompressedMessageBytes);
+ return this;
+ }
+
public LogConfigBuilder indexIntervalBytes(int indexIntervalBytes) {
configs.put(TopicConfig.INDEX_INTERVAL_BYTES_CONFIG, indexIntervalBytes);
return this;
diff --git a/storage/src/test/java/org/apache/kafka/storage/internals/log/LogValidatorTest.java b/storage/src/test/java/org/apache/kafka/storage/internals/log/LogValidatorTest.java
index d21ec04e0a4..4c806551312 100644
--- a/storage/src/test/java/org/apache/kafka/storage/internals/log/LogValidatorTest.java
+++ b/storage/src/test/java/org/apache/kafka/storage/internals/log/LogValidatorTest.java
@@ -188,6 +188,51 @@ public class LogValidatorTest {
));
}
+ /**
+ * The compressed validation path enforces the per-record decompressed-body-size limit
+ * (max.decompressed.message.bytes): a record whose decompressed body exceeds the configured
+ * limit is rejected with InvalidRecordException before the body is allocated, while the
+ * limit-less constructor applies no effective limit.
+ */
+ @Test
+ public void testCompressedRecordExceedingMaxRecordBodySizeIsRejected() {
+ MemoryRecords records = MemoryRecords.withRecords(Compression.gzip().build(),
+ new SimpleRecord(System.currentTimeMillis(), "key".getBytes(), new byte[1000]));
+
+ new LogValidator(records,
+ topicPartition,
+ time,
+ CompressionType.GZIP,
+ Compression.gzip().build(),
+ false,
+ RecordBatch.CURRENT_MAGIC_VALUE,
+ TimestampType.CREATE_TIME,
+ 5000L,
+ 5000L,
+ RecordBatch.NO_PARTITION_LEADER_EPOCH,
+ AppendOrigin.CLIENT
+ ).validateMessagesAndAssignOffsets(
+ PrimitiveRef.ofLong(0), metricsRecorder, RequestLocal.withThreadConfinedCaching().bufferSupplier());
+
+ InvalidRecordException e = assertThrows(InvalidRecordException.class, () -> new LogValidator(records,
+ topicPartition,
+ time,
+ CompressionType.GZIP,
+ Compression.gzip().build(),
+ false,
+ RecordBatch.CURRENT_MAGIC_VALUE,
+ TimestampType.CREATE_TIME,
+ 5000L,
+ 5000L,
+ RecordBatch.NO_PARTITION_LEADER_EPOCH,
+ AppendOrigin.CLIENT,
+ 100
+ ).validateMessagesAndAssignOffsets(
+ PrimitiveRef.ofLong(0), metricsRecorder, RequestLocal.withThreadConfinedCaching().bufferSupplier()));
+ assertTrue(e.getMessage().contains("exceeds the configured maximum record size"),
+ "expected the configured-maximum guard, got: " + e.getMessage());
+ }
+
@ParameterizedTest
@CsvSource({"0,1,gzip", "1,0,gzip"})
public void checkMismatchMagic(byte batchMagic, byte recordMagic, String compressionName) {
diff --git a/storage/src/test/java/org/apache/kafka/storage/internals/log/UnifiedLogTest.java b/storage/src/test/java/org/apache/kafka/storage/internals/log/UnifiedLogTest.java
index 4fa95365843..bb4ba94246b 100644
--- a/storage/src/test/java/org/apache/kafka/storage/internals/log/UnifiedLogTest.java
+++ b/storage/src/test/java/org/apache/kafka/storage/internals/log/UnifiedLogTest.java
@@ -1509,7 +1509,7 @@ public class UnifiedLogTest {
public boolean shouldRetainRecord(RecordBatch recordBatch, Record record) {
return !record.hasKey();
}
- }, filtered, BufferSupplier.NO_CACHING);
+ }, filtered, BufferSupplier.NO_CACHING, Records.SOFT_MAX_ARRAY_LENGTH);
filtered.flip();
MemoryRecords filteredRecords = MemoryRecords.readableRecords(filtered);
@@ -1561,7 +1561,7 @@ public class UnifiedLogTest {
@Override public boolean shouldRetainRecord(RecordBatch recordBatch, Record record) {
return false;
}
- }, filtered, BufferSupplier.NO_CACHING);
+ }, filtered, BufferSupplier.NO_CACHING, Records.SOFT_MAX_ARRAY_LENGTH);
filtered.flip();
MemoryRecords filteredRecords = MemoryRecords.readableRecords(filtered);
@@ -1614,7 +1614,7 @@ public class UnifiedLogTest {
@Override public boolean shouldRetainRecord(RecordBatch recordBatch, Record record) {
return !record.hasKey();
}
- }, filtered, BufferSupplier.NO_CACHING);
+ }, filtered, BufferSupplier.NO_CACHING, Records.SOFT_MAX_ARRAY_LENGTH);
filtered.flip();
MemoryRecords filteredRecords = MemoryRecords.readableRecords(filtered);
@@ -3085,6 +3085,37 @@ public class UnifiedLogTest {
log.appendAsLeader(messageSetWithCompressedKeyedMessage, 0);
}
+ /**
+ * End-to-end produce-path enforcement of the per-record decompressed-body-size limit
+ * (max.decompressed.message.bytes): a record that is tiny gzip-compressed on the wire (well
+ * under max.message.bytes) but whose declared decompressed body exceeds the configured limit is
+ * rejected as an invalid record before the body is allocated. A small compressed record and an
+ * equally-large uncompressed record are unaffected.
+ */
+ @Test
+ public void testAppendCompressedRecordExceedingMaxDecompressedMessageBytesIsRejected() throws IOException {
+ LogConfig logConfig = new LogTestUtils.LogConfigBuilder()
+ .maxDecompressedMessageBytes(100)
+ .build();
+ log = createLog(logDir, logConfig);
+
+ MemoryRecords oversizedCompressed = MemoryRecords.withRecords(Compression.gzip().build(),
+ new SimpleRecord("key".getBytes(), new byte[1000]));
+ InvalidRecordException e = assertThrows(InvalidRecordException.class,
+ () -> log.appendAsLeader(oversizedCompressed, 0));
+ assertTrue(e.getMessage().contains("exceeds the configured maximum record size"),
+ "expected the configured-maximum guard, got: " + e.getMessage());
+ assertEquals(0, log.logEndOffset(), "a rejected record must not be appended");
+
+ // The limit bounds only the decompressed per-record body; uncompressed records are bounded
+ // on the wire by max.message.bytes.
+ log.appendAsLeader(MemoryRecords.withRecords(Compression.gzip().build(),
+ new SimpleRecord("key".getBytes(), new byte[64])), 0);
+ log.appendAsLeader(MemoryRecords.withRecords(Compression.NONE,
+ new SimpleRecord("key".getBytes(), new byte[1000])), 0);
+ assertEquals(2, log.logEndOffset());
+ }
+
/**
* We have a max size limit on message appends, check that it is properly enforced by appending a message larger than the
* setting and checking that an exception is thrown.
@@ -6055,4 +6086,53 @@ public class UnifiedLogTest {
assertTrue(exception.getMessage().contains(String.valueOf(originalEpoch)));
assertTrue(exception.getMessage().contains(String.valueOf(bumpedEpoch)));
}
+
+
+ @Test
+ public void testFetchOffsetByTimestampRejectsCompressedRecordExceedingMaxDecompressedMessageBytes() throws IOException {
+ LogConfig logConfig = new LogTestUtils.LogConfigBuilder()
+ .maxDecompressedMessageBytes(100)
+ .build();
+ log = createLog(logDir, logConfig);
+ long firstTimestamp = mockTime.milliseconds();
+ long secondTimestamp = firstTimestamp + 1;
+ Compression gzip = Compression.gzip().build();
+ // appendAsFollower bypasses produce validation, so the oversized record becomes durable
+ log.appendAsFollower(MemoryRecords.withRecords(0L, gzip, 0,
+ new SimpleRecord(firstTimestamp, "key".getBytes(), new byte[10])), 0);
+ log.appendAsFollower(MemoryRecords.withRecords(1L, gzip, 0,
+ new SimpleRecord(secondTimestamp, "key".getBytes(), new byte[1000])), 0);
+
+ // A lookup that only decompresses the small record succeeds
+ assertEquals(new OffsetResultHolder(new FileRecords.TimestampAndOffset(firstTimestamp, 0L, Optional.of(0))),
+ log.fetchOffsetByTimestamp(firstTimestamp, Optional.empty()));
+ // A lookup that has to decompress the oversized record is rejected before its body is allocated
+ InvalidRecordException e = assertThrows(InvalidRecordException.class,
+ () -> log.fetchOffsetByTimestamp(secondTimestamp, Optional.empty()));
+ assertTrue(e.getMessage().contains("exceeds the configured maximum record size of 100"), e.getMessage());
+ }
+
+ @Test
+ public void testFetchOffsetByMaxTimestampRejectsCompressedRecordExceedingMaxDecompressedMessageBytes() throws IOException {
+ LogConfig logConfig = new LogTestUtils.LogConfigBuilder()
+ .maxDecompressedMessageBytes(100)
+ .build();
+ log = createLog(logDir, logConfig);
+ long firstTimestamp = mockTime.milliseconds();
+ Compression gzip = Compression.gzip().build();
+ // appendAsFollower bypasses produce validation, so the oversized records become durable
+ log.appendAsFollower(MemoryRecords.withRecords(0L, gzip, 0,
+ new SimpleRecord(firstTimestamp, "key".getBytes(), new byte[1000])), 0);
+ log.appendAsFollower(MemoryRecords.withRecords(1L, gzip, 0,
+ new SimpleRecord(firstTimestamp + 1, "key".getBytes(), new byte[10])), 0);
+ // Resolving MAX_TIMESTAMP only decompresses the batch holding the max timestamp, here the small one
+ assertEquals(new OffsetResultHolder(new FileRecords.TimestampAndOffset(firstTimestamp + 1, 1L, Optional.of(0))),
+ log.fetchOffsetByTimestamp(ListOffsetsRequest.MAX_TIMESTAMP, Optional.empty()));
+ // Once the oversized batch holds the max timestamp, the lookup is rejected before its body is allocated
+ log.appendAsFollower(MemoryRecords.withRecords(2L, gzip, 0,
+ new SimpleRecord(firstTimestamp + 2, "key".getBytes(), new byte[1000])), 0);
+ InvalidRecordException e = assertThrows(InvalidRecordException.class,
+ () -> log.fetchOffsetByTimestamp(ListOffsetsRequest.MAX_TIMESTAMP, Optional.empty()));
+ assertTrue(e.getMessage().contains("exceeds the configured maximum record size of 100"), e.getMessage());
+ }
}