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 12:11:27 UTC
(kafka) branch 4.2 updated: Bound decompressed record size (#23449)
This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch 4.2
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/4.2 by this push:
new 62e667d6178 Bound decompressed record size (#23449)
62e667d6178 is described below
commit 62e667d617819226f747f5c1c63f6325c6ba5dbe
Author: Manikumar Reddy <ma...@gmail.com>
AuthorDate: Sun Sep 13 17:41:20 2026 +0530
Bound decompressed record size (#23449)
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)
(cherry picked from commit a544dc8effd8f1bf90c5746130c91b2a8127e9db)
(cherry picked from commit c9cea9f564195771e72f1c9a2c68059bb044e599)
Co-authored-by: Srinivas Akhil Mallela <sr...@gmail.com>
---
checkstyle/suppressions.xml | 4 +-
.../apache/kafka/common/config/TopicConfig.java | 12 ++
.../common/record/AbstractLegacyRecordBatch.java | 36 ++++-
.../apache/kafka/common/record/DefaultRecord.java | 34 ++++-
.../kafka/common/record/DefaultRecordBatch.java | 22 ++-
.../kafka/common/record/FileLogInputStream.java | 5 +
.../apache/kafka/common/record/FileRecords.java | 22 ++-
.../apache/kafka/common/record/MemoryRecords.java | 28 ++--
.../kafka/common/record/MutableRecordBatch.java | 9 ++
.../apache/kafka/common/record/RecordBatch.java | 16 +-
.../org/apache/kafka/common/record/Records.java | 7 +
.../internals/FetchRequestManagerTest.java | 2 +-
.../clients/consumer/internals/FetcherTest.java | 2 +-
.../internals/ShareConsumeRequestManagerTest.java | 3 +-
.../record/AbstractLegacyRecordBatchTest.java | 65 ++++++++
.../common/record/DefaultRecordBatchTest.java | 95 ++++++++++++
.../kafka/common/record/DefaultRecordTest.java | 115 ++++++++++++++-
.../kafka/common/record/FileRecordsTest.java | 28 +++-
.../kafka/common/record/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/LogCleanerTest.scala | 28 ++++
.../test/scala/unit/kafka/log/LogConfigTest.scala | 21 +++
.../test/scala/unit/kafka/log/LogTestUtils.scala | 4 +-
.../test/scala/unit/kafka/log/UnifiedLogTest.scala | 76 +++++++++-
.../scala/unit/kafka/server/KafkaApisTest.scala | 34 ++++-
.../scala/unit/kafka/server/KafkaConfigTest.scala | 2 +
.../unit/kafka/server/ProduceRequestTest.scala | 163 ++++++++++++++++++++-
.../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 ++-
.../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 | 43 +++++-
.../storage/internals/log/LogSegmentTest.java | 18 +--
.../storage/internals/log/LogValidatorTest.java | 45 ++++++
41 files changed, 982 insertions(+), 99 deletions(-)
diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml
index ff60f0d0fbb..703fea90c5d 100644
--- a/checkstyle/suppressions.xml
+++ b/checkstyle/suppressions.xml
@@ -362,9 +362,11 @@
files="(LocalLog|LogLoader|LogValidator|RemoteLogManager|RemoteIndexCache|UnifiedLog).java"/>
<suppress checks="ParameterNumber"
files="(LogAppendInfo|LogLoader|RemoteLogManagerConfig|UnifiedLog).java"/>
+ <suppress checks="ParameterNumber"
+ files="[/\\]Cleaner\.java"/>
<suppress checks="(ClassDataAbstractionCoupling|ClassFanOutComplexity)"
files="(UnifiedLog|RemoteLogManager|RemoteLogManagerTest).java"/>
- <suppress checks="MethodLength" files="(RemoteLogManager|RemoteLogManagerConfig).java"/>
+ <suppress checks="MethodLength" files="(RemoteLogManager|RemoteLogManagerConfig|UnifiedLog).java"/>
<suppress checks="JavaNCSS" files="RemoteLogManagerTest.java"/>
<!-- benchmarks -->
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 e97c39bc619..8d18b666f14 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
@@ -110,6 +110,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/AbstractLegacyRecordBatch.java b/clients/src/main/java/org/apache/kafka/common/record/AbstractLegacyRecordBatch.java
index e47d7c866cc..5d1f024811e 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/AbstractLegacyRecordBatch.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/AbstractLegacyRecordBatch.java
@@ -232,8 +232,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;
@@ -267,6 +271,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);
@@ -280,11 +290,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);
}
@@ -300,6 +312,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);
@@ -319,7 +340,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)
@@ -334,7 +356,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();
@@ -517,6 +539,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/DefaultRecord.java b/clients/src/main/java/org/apache/kafka/common/record/DefaultRecord.java
index 783155a0061..9a302cc0bc3 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/DefaultRecord.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/DefaultRecord.java
@@ -276,12 +276,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)
@@ -358,12 +374,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/DefaultRecordBatch.java b/clients/src/main/java/org/apache/kafka/common/record/DefaultRecordBatch.java
index d6e9cc6bd7f..9c923938095 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/DefaultRecordBatch.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/DefaultRecordBatch.java
@@ -275,21 +275,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);
}
};
}
@@ -327,7 +327,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());
@@ -337,6 +337,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());
}
@@ -351,13 +356,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/FileLogInputStream.java b/clients/src/main/java/org/apache/kafka/common/record/FileLogInputStream.java
index bd966e79f03..358a9761b59 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/FileLogInputStream.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/FileLogInputStream.java
@@ -158,6 +158,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/FileRecords.java b/clients/src/main/java/org/apache/kafka/common/record/FileRecords.java
index 2f5e2e50dde..795a1f53e65 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/FileRecords.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/FileRecords.java
@@ -20,6 +20,8 @@ import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.network.TransferableChannel;
import org.apache.kafka.common.record.FileLogInputStream.FileChannelRecordBatch;
import org.apache.kafka.common.utils.AbstractIterator;
+import org.apache.kafka.common.utils.BufferSupplier;
+import org.apache.kafka.common.utils.CloseableIterator;
import org.apache.kafka.common.utils.Utils;
import java.io.Closeable;
@@ -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/MemoryRecords.java b/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java
index f806cce6920..32b3a3e700a 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java
@@ -126,6 +126,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
@@ -133,21 +136,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;
@@ -158,7 +157,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;
@@ -215,8 +214,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/MutableRecordBatch.java b/clients/src/main/java/org/apache/kafka/common/record/MutableRecordBatch.java
index b5f42e5b915..ba4e6aa19af 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/MutableRecordBatch.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/MutableRecordBatch.java
@@ -65,4 +65,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/RecordBatch.java b/clients/src/main/java/org/apache/kafka/common/record/RecordBatch.java
index e36beff08f2..87b0e13f063 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/RecordBatch.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/RecordBatch.java
@@ -239,6 +239,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.
@@ -252,12 +261,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/Records.java b/clients/src/main/java/org/apache/kafka/common/record/Records.java
index 017c49ba94c..b6e2d294a5e 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/Records.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/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 f8e53148b01..d6cbbcb2f68 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
@@ -2549,7 +2549,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 93da0a4433a..4e1530fb3c1 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
@@ -2536,7 +2536,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 b6040950ccd..797074fcc04 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
@@ -59,6 +59,7 @@ import org.apache.kafka.common.record.MemoryRecords;
import org.apache.kafka.common.record.MemoryRecordsBuilder;
import org.apache.kafka.common.record.Record;
import org.apache.kafka.common.record.RecordBatch;
+import org.apache.kafka.common.record.Records;
import org.apache.kafka.common.record.SimpleRecord;
import org.apache.kafka.common.record.TimestampType;
import org.apache.kafka.common.requests.MetadataResponse;
@@ -1783,7 +1784,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/AbstractLegacyRecordBatchTest.java b/clients/src/test/java/org/apache/kafka/common/record/AbstractLegacyRecordBatchTest.java
index 015cebec1b0..f741c6f5b8d 100644
--- a/clients/src/test/java/org/apache/kafka/common/record/AbstractLegacyRecordBatchTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/record/AbstractLegacyRecordBatchTest.java
@@ -19,10 +19,16 @@ package org.apache.kafka.common.record;
import org.apache.kafka.common.InvalidRecordException;
import org.apache.kafka.common.compress.Compression;
import org.apache.kafka.common.record.AbstractLegacyRecordBatch.ByteBufferLegacyRecordBatch;
+import org.apache.kafka.common.utils.BufferSupplier;
+import org.apache.kafka.common.utils.ByteBufferOutputStream;
+import org.apache.kafka.common.utils.CloseableIterator;
import org.apache.kafka.common.utils.Utils;
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;
@@ -250,4 +256,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/DefaultRecordBatchTest.java b/clients/src/test/java/org/apache/kafka/common/record/DefaultRecordBatchTest.java
index e63dafc3a0d..b193e405988 100644
--- a/clients/src/test/java/org/apache/kafka/common/record/DefaultRecordBatchTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/record/DefaultRecordBatchTest.java
@@ -23,6 +23,8 @@ import org.apache.kafka.common.errors.CorruptRecordException;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.common.utils.BufferSupplier;
+import org.apache.kafka.common.utils.ByteBufferOutputStream;
+import org.apache.kafka.common.utils.ByteUtils;
import org.apache.kafka.common.utils.ChunkedBytesStream;
import org.apache.kafka.common.utils.CloseableIterator;
import org.apache.kafka.common.utils.Utils;
@@ -34,11 +36,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;
@@ -566,6 +570,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/DefaultRecordTest.java b/clients/src/test/java/org/apache/kafka/common/record/DefaultRecordTest.java
index 2473cca5410..608d95f030f 100644
--- a/clients/src/test/java/org/apache/kafka/common/record/DefaultRecordTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/record/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/FileRecordsTest.java b/clients/src/test/java/org/apache/kafka/common/record/FileRecordsTest.java
index a9d12285f12..cb91b90f437 100644
--- a/clients/src/test/java/org/apache/kafka/common/record/FileRecordsTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/record/FileRecordsTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.common.record;
+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;
@@ -457,12 +458,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,
@@ -789,4 +790,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/MemoryRecordsTest.java b/clients/src/test/java/org/apache/kafka/common/record/MemoryRecordsTest.java
index 7092928010b..2b573e9788c 100644
--- a/clients/src/test/java/org/apache/kafka/common/record/MemoryRecordsTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/record/MemoryRecordsTest.java
@@ -310,7 +310,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);
@@ -363,7 +363,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());
@@ -424,7 +424,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());
@@ -471,7 +471,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());
@@ -547,7 +547,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);
@@ -649,7 +649,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);
@@ -685,7 +685,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);
@@ -760,7 +760,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);
@@ -852,7 +852,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();
@@ -899,7 +899,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();
@@ -1017,7 +1017,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 743937b54fc..a26d7281b63 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 961ec31acbf..8d32711bba3 100755
--- a/core/src/main/scala/kafka/server/KafkaConfig.scala
+++ b/core/src/main/scala/kafka/server/KafkaConfig.scala
@@ -249,6 +249,8 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _], enforceProv
val connectionSetupTimeoutMs = getLong(ServerConfigs.SOCKET_CONNECTION_SETUP_TIMEOUT_MS_CONFIG)
val connectionSetupTimeoutMaxMs = getLong(ServerConfigs.SOCKET_CONNECTION_SETUP_TIMEOUT_MAX_MS_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
@@ -704,6 +706,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/LogCleanerTest.scala b/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
index 318a0c66140..7905ded58b9 100644
--- a/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
@@ -19,6 +19,7 @@ package kafka.log
import kafka.server.KafkaConfig
import kafka.utils.{CoreUtils, Logging, TestUtils}
+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.TopicConfig
@@ -991,6 +992,33 @@ class LogCleanerTest extends Logging {
)
}
+ /**
+ * 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
+ def testCleanRejectsRecordExceedingConfiguredMaxDecompressedMessageBytes(): Unit = {
+ val cleaner = makeCleaner(10)
+ val logProps = new Properties()
+ logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT)
+ logProps.put(TopicConfig.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG, "10")
+ val log = makeLog(config = LogConfig.fromProps(logConfig.originals, logProps))
+
+ // partitionLeaderEpoch must be valid (>= 0) for follower append to assign it to the epoch cache.
+ val 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 Array[Byte](100)))
+ log.appendAsFollower(oversized, Int.MaxValue)
+ log.roll()
+
+ val e = assertThrows(classOf[InvalidRecordException],
+ () => cleaner.clean(new LogToClean(log, 0L, log.activeSegment.baseOffset, false)))
+ assertTrue(e.getMessage.contains("exceeds the configured maximum record size"),
+ s"expected the configured-maximum guard message, got: ${e.getMessage}")
+ }
+
def createLogWithMessagesLargerThanMaxSize(largeMessageSize: Int): (UnifiedLog, FakeOffsetMap) = {
val logProps = new Properties()
logProps.put(LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG, largeMessageSize * 16: java.lang.Integer)
diff --git a/core/src/test/scala/unit/kafka/log/LogConfigTest.scala b/core/src/test/scala/unit/kafka/log/LogConfigTest.scala
index 98a49070913..4ccb6c09662 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.Records
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.Test
@@ -79,6 +80,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 Properties
diff --git a/core/src/test/scala/unit/kafka/log/LogTestUtils.scala b/core/src/test/scala/unit/kafka/log/LogTestUtils.scala
index 1881b3629ab..acfd7ae8b61 100644
--- a/core/src/test/scala/unit/kafka/log/LogTestUtils.scala
+++ b/core/src/test/scala/unit/kafka/log/LogTestUtils.scala
@@ -73,7 +73,8 @@ object LogTestUtils {
fileDeleteDelayMs: Long = ServerLogConfigs.LOG_DELETE_DELAY_MS_DEFAULT,
remoteLogStorageEnable: Boolean = LogConfig.DEFAULT_REMOTE_STORAGE_ENABLE,
remoteLogCopyDisable: Boolean = DEFAULT_REMOTE_LOG_COPY_DISABLE_CONFIG,
- remoteLogDeleteOnDisable: Boolean = DEFAULT_REMOTE_LOG_DELETE_ON_DISABLE_CONFIG): LogConfig = {
+ remoteLogDeleteOnDisable: Boolean = DEFAULT_REMOTE_LOG_DELETE_ON_DISABLE_CONFIG,
+ maxDecompressedMessageBytes: Int = ServerLogConfigs.MAX_DECOMPRESSED_MESSAGE_BYTES_DEFAULT): LogConfig = {
val logProps = new Properties()
logProps.put(TopicConfig.SEGMENT_MS_CONFIG, segmentMs: java.lang.Long)
logProps.put(LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG, segmentBytes: Integer)
@@ -90,6 +91,7 @@ object LogTestUtils {
logProps.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, remoteLogStorageEnable: java.lang.Boolean)
logProps.put(TopicConfig.REMOTE_LOG_COPY_DISABLE_CONFIG, remoteLogCopyDisable: java.lang.Boolean)
logProps.put(TopicConfig.REMOTE_LOG_DELETE_ON_DISABLE_CONFIG, remoteLogDeleteOnDisable: java.lang.Boolean)
+ logProps.put(TopicConfig.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG, maxDecompressedMessageBytes: Integer)
new LogConfig(logProps)
}
diff --git a/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala b/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala
index 0155c1008c6..ab211be5fbe 100755
--- a/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala
+++ b/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala
@@ -822,7 +822,7 @@ class UnifiedLogTest {
override def checkBatchRetention(batch: RecordBatch): RecordFilter.BatchRetentionResult =
new RecordFilter.BatchRetentionResult(RecordFilter.BatchRetention.DELETE_EMPTY, false)
override def shouldRetainRecord(recordBatch: RecordBatch, record: Record): Boolean = !record.hasKey
- }, filtered, BufferSupplier.NO_CACHING)
+ }, filtered, BufferSupplier.NO_CACHING, Records.SOFT_MAX_ARRAY_LENGTH)
filtered.flip()
val filteredRecords = MemoryRecords.readableRecords(filtered)
@@ -876,7 +876,7 @@ class UnifiedLogTest {
override def checkBatchRetention(batch: RecordBatch): RecordFilter.BatchRetentionResult =
new RecordFilter.BatchRetentionResult(RecordFilter.BatchRetention.RETAIN_EMPTY, true)
override def shouldRetainRecord(recordBatch: RecordBatch, record: Record): Boolean = false
- }, filtered, BufferSupplier.NO_CACHING)
+ }, filtered, BufferSupplier.NO_CACHING, Records.SOFT_MAX_ARRAY_LENGTH)
filtered.flip()
val filteredRecords = MemoryRecords.readableRecords(filtered)
@@ -932,7 +932,7 @@ class UnifiedLogTest {
override def checkBatchRetention(batch: RecordBatch): RecordFilter.BatchRetentionResult =
new RecordFilter.BatchRetentionResult(RecordFilter.BatchRetention.DELETE_EMPTY, false)
override def shouldRetainRecord(recordBatch: RecordBatch, record: Record): Boolean = !record.hasKey
- }, filtered, BufferSupplier.NO_CACHING)
+ }, filtered, BufferSupplier.NO_CACHING, Records.SOFT_MAX_ARRAY_LENGTH)
filtered.flip()
val filteredRecords = MemoryRecords.readableRecords(filtered)
@@ -2199,6 +2199,76 @@ class UnifiedLogTest {
log.fetchOffsetByTimestamp(ListOffsetsRequest.LATEST_TIMESTAMP, Optional.empty))
}
+ /**
+ * 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
+ def testAppendCompressedRecordExceedingMaxDecompressedMessageBytesIsRejected(): Unit = {
+ val logConfig = LogTestUtils.createLogConfig(maxDecompressedMessageBytes = 100)
+ val log = createLog(logDir, logConfig)
+
+ val oversizedCompressed = MemoryRecords.withRecords(Compression.gzip().build(),
+ new SimpleRecord("key".getBytes, new Array[Byte](1000)))
+ val e = assertThrows(classOf[InvalidRecordException], () => log.appendAsLeader(oversizedCompressed, 0))
+ assertTrue(e.getMessage.contains("exceeds the configured maximum record size"),
+ s"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 Array[Byte](64))), 0)
+ log.appendAsLeader(MemoryRecords.withRecords(Compression.NONE,
+ new SimpleRecord("key".getBytes, new Array[Byte](1000))), 0)
+ assertEquals(2, log.logEndOffset)
+ }
+
+ @Test
+ def testFetchOffsetByTimestampRejectsCompressedRecordExceedingMaxDecompressedMessageBytes(): Unit = {
+ val logConfig = LogTestUtils.createLogConfig(maxDecompressedMessageBytes = 100)
+ val log = createLog(logDir, logConfig)
+ val firstTimestamp = mockTime.milliseconds
+ val secondTimestamp = firstTimestamp + 1
+ val 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 Array[Byte](10))), 0)
+ log.appendAsFollower(MemoryRecords.withRecords(1L, gzip, 0,
+ new SimpleRecord(secondTimestamp, "key".getBytes, new Array[Byte](1000))), 0)
+
+ // A lookup that only decompresses the small record succeeds
+ assertEquals(new OffsetResultHolder(new 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
+ val e = assertThrows(classOf[InvalidRecordException], () => log.fetchOffsetByTimestamp(secondTimestamp, Optional.empty))
+ assertTrue(e.getMessage.contains("exceeds the configured maximum record size of 100"), e.getMessage)
+ }
+
+ @Test
+ def testFetchOffsetByMaxTimestampRejectsCompressedRecordExceedingMaxDecompressedMessageBytes(): Unit = {
+ val logConfig = LogTestUtils.createLogConfig(maxDecompressedMessageBytes = 100)
+ val log = createLog(logDir, logConfig)
+ val firstTimestamp = mockTime.milliseconds
+ val 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 Array[Byte](1000))), 0)
+ log.appendAsFollower(MemoryRecords.withRecords(1L, gzip, 0,
+ new SimpleRecord(firstTimestamp + 1, "key".getBytes, new Array[Byte](10))), 0)
+ // Resolving MAX_TIMESTAMP only decompresses the batch holding the max timestamp, here the small one
+ assertEquals(new OffsetResultHolder(new 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 Array[Byte](1000))), 0)
+ val e = assertThrows(classOf[InvalidRecordException], () => log.fetchOffsetByTimestamp(ListOffsetsRequest.MAX_TIMESTAMP, Optional.empty))
+ assertTrue(e.getMessage.contains("exceeds the configured maximum record size of 100"), e.getMessage)
+ }
+
@Test
def testFetchOffsetByTimestampWithMaxTimestampIncludesTimestamp(): Unit = {
val logConfig = LogTestUtils.createLogConfig(segmentBytes = 200, indexIntervalBytes = 1)
diff --git a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
index a20139cf23c..14ba3578637 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}
@@ -306,6 +306,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 99565e90084..65bbebf3d81 100755
--- a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
@@ -1198,6 +1198,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/core/src/test/scala/unit/kafka/server/ProduceRequestTest.scala b/core/src/test/scala/unit/kafka/server/ProduceRequestTest.scala
index 57545c7ba2b..3a6573e4123 100644
--- a/core/src/test/scala/unit/kafka/server/ProduceRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ProduceRequestTest.scala
@@ -20,11 +20,11 @@ package kafka.server
import java.nio.ByteBuffer
import java.util.{Collections, Properties}
import kafka.utils.TestUtils
-import org.apache.kafka.clients.admin.{Admin, TopicDescription}
-import org.apache.kafka.common.{TopicIdPartition, TopicPartition}
+import org.apache.kafka.clients.admin.{Admin, AlterConfigOp, ConfigEntry, TopicDescription}
+import org.apache.kafka.common.{TopicIdPartition, TopicPartition, Uuid}
import org.apache.kafka.common.compress.Compression
-import org.apache.kafka.common.config.TopicConfig
-import org.apache.kafka.common.message.ProduceRequestData
+import org.apache.kafka.common.config.{ConfigResource, TopicConfig}
+import org.apache.kafka.common.message.{ProduceRequestData, ProduceResponseData}
import org.apache.kafka.common.protocol.{ApiKeys, Errors}
import org.apache.kafka.common.record._
import org.apache.kafka.common.requests.{ProduceRequest, ProduceResponse}
@@ -282,6 +282,161 @@ class ProduceRequestTest extends BaseRequestTest {
assertEquals(-1, partitionProduceResponse1.logAppendTimeMs)
}
+ private val 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 val OVERSIZED_VALUE_BYTES = 4096
+ private val 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.
+ */
+ @Test
+ def testProduceRejectsCompressedRecordExceedingMaxDecompressedMessageBytes(): Unit = {
+ val topic = "topic"
+ val topicConfig = new Properties
+ topicConfig.setProperty(TopicConfig.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG, SMALL_MAX_DECOMPRESSED_MESSAGE_BYTES)
+ val partitionToLeader = createTopic(topic, topicConfig = topicConfig)
+ val leader = partitionToLeader(0)
+ val topicId = getTopicIds().get(topic).get
+
+ val rejected = onlyPartitionResponse(sendProduceRequest(leader,
+ 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(-1, rejected.baseOffset, "a rejected record must not be appended")
+
+ val acceptedSmall = onlyPartitionResponse(sendProduceRequest(leader,
+ 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")
+
+ val acceptedUncompressed = onlyPartitionResponse(sendProduceRequest(leader,
+ 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.
+ */
+ @Test
+ def testMaxDecompressedMessageBytesIsDynamicallyReconfigurable(): Unit = {
+ val topic = "topic"
+ val partitionToLeader = createTopic(topic)
+ val leader = partitionToLeader(0)
+ val topicId = getTopicIds().get(topic).get
+
+ val before = onlyPartitionResponse(sendProduceRequest(leader,
+ 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")
+
+ val admin = createAdminClient()
+ val resource = new ConfigResource(ConfigResource.Type.TOPIC, topic)
+ admin.incrementalAlterConfigs(Map(resource -> List(new AlterConfigOp(
+ new ConfigEntry(TopicConfig.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG, SMALL_MAX_DECOMPRESSED_MESSAGE_BYTES),
+ AlterConfigOp.OpType.SET)).asJavaCollection).asJava).all.get
+
+ // The topic-config change reaches the produce path asynchronously; poll until it takes effect.
+ TestUtils.waitUntilTrue(
+ () => onlyPartitionResponse(sendProduceRequest(leader,
+ produceRequest(topicId, singleRecord(Compression.gzip().build(), OVERSIZED_VALUE_BYTES))))
+ .errorCode == Errors.INVALID_RECORD.code,
+ s"the lowered topic-level ${TopicConfig.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG} was not applied to the produce path",
+ 15000L)
+ }
+
+ /**
+ * A broker-level default is inherited by a topic without an override: an oversized compressed
+ * record is rejected. The default is lowered dynamically (via a BROKER-resource alter) rather
+ * than at cluster startup, since this test suite shares one cluster per test method with no
+ * per-test static broker-config override.
+ */
+ @Test
+ def testBrokerDefaultMaxDecompressedMessageBytesAppliesToTopicWithoutOverride(): Unit = {
+ val topic = "topic"
+ val admin = createAdminClient()
+ // Empty resource name = cluster-wide broker default.
+ val resource = new ConfigResource(ConfigResource.Type.BROKER, "")
+ admin.incrementalAlterConfigs(Map(resource -> List(new AlterConfigOp(
+ new ConfigEntry(TopicConfig.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG, SMALL_MAX_DECOMPRESSED_MESSAGE_BYTES),
+ AlterConfigOp.OpType.SET)).asJavaCollection).asJava).all.get
+
+ val partitionToLeader = createTopic(topic)
+ val leader = partitionToLeader(0)
+ val topicId = getTopicIds().get(topic).get
+
+ // The broker-default change reaches the produce path asynchronously; poll until it takes effect.
+ TestUtils.waitUntilTrue(
+ () => onlyPartitionResponse(sendProduceRequest(leader,
+ produceRequest(topicId, singleRecord(Compression.gzip().build(), OVERSIZED_VALUE_BYTES))))
+ .errorCode == Errors.INVALID_RECORD.code,
+ "the broker-level default per-record limit was not applied to a topic without an override",
+ 15000L)
+ }
+
+ /**
+ * 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 -- no restart.
+ */
+ @Test
+ def testBrokerDefaultMaxDecompressedMessageBytesIsDynamicallyReconfigurable(): Unit = {
+ val topic = "topic"
+ val partitionToLeader = createTopic(topic)
+ val leader = partitionToLeader(0)
+ val topicId = getTopicIds().get(topic).get
+
+ val before = onlyPartitionResponse(sendProduceRequest(leader,
+ 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")
+
+ val admin = createAdminClient()
+ // Empty resource name = cluster-wide broker default.
+ val resource = new ConfigResource(ConfigResource.Type.BROKER, "")
+ admin.incrementalAlterConfigs(Map(resource -> List(new AlterConfigOp(
+ new ConfigEntry(TopicConfig.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG, SMALL_MAX_DECOMPRESSED_MESSAGE_BYTES),
+ AlterConfigOp.OpType.SET)).asJavaCollection).asJava).all.get
+
+ // The broker-default change reaches the produce path asynchronously; poll until it takes effect.
+ TestUtils.waitUntilTrue(
+ () => onlyPartitionResponse(sendProduceRequest(leader,
+ produceRequest(topicId, singleRecord(Compression.gzip().build(), OVERSIZED_VALUE_BYTES))))
+ .errorCode == Errors.INVALID_RECORD.code,
+ s"the lowered cluster-wide broker-default ${TopicConfig.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG} was not applied to the produce path",
+ 15000L)
+ }
+
+ private def produceRequest(topicId: Uuid, records: MemoryRecords): ProduceRequest = {
+ 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((-1).toShort)
+ .setTimeoutMs(3000)
+ .setTransactionalId(null)).build()
+ }
+
+ private def singleRecord(compression: Compression, valueSize: Int): MemoryRecords = {
+ MemoryRecords.withRecords(compression,
+ new SimpleRecord(System.currentTimeMillis(), "key".getBytes, new Array[Byte](valueSize)))
+ }
+
+ private def onlyPartitionResponse(response: ProduceResponse): ProduceResponseData.PartitionProduceResponse = {
+ assertEquals(1, response.data.responses.size)
+ val topicProduceResponse = response.data.responses.asScala.head
+ assertEquals(1, topicProduceResponse.partitionResponses.size)
+ topicProduceResponse.partitionResponses.asScala.head
+ }
+
private def sendProduceRequest(leaderId: Int, request: ProduceRequest): ProduceResponse = {
connectAndReceive[ProduceResponse](request, destination = brokerSocketServer(leaderId))
}
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 e40c5c8e6d3..4dfebcd4d89 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.CompressionType;
+import org.apache.kafka.common.record.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;
@@ -46,6 +48,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;
@@ -128,6 +133,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 5438a0a59cb..37db03f185e 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
@@ -159,5 +159,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 c05f9f2816a..1145ad15b4c 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/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 a9848633eff..d1d90a605eb 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
@@ -629,8 +629,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;
@@ -643,7 +643,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)
@@ -717,6 +717,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);
@@ -737,12 +738,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 e17809d988e..00a2345df1d 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
@@ -252,7 +252,8 @@ public class Cleaner {
lastOffsetOfActiveProducers,
upperBoundOffsetOfCleaningRound,
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
@@ -302,6 +303,8 @@ public class Cleaner {
* @param upperBoundOffsetOfCleaningRound Next offset of the last batch in the source segment
* @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,
@@ -314,7 +317,8 @@ public class Cleaner {
Map<Long, LastRecord> lastRecordsOfActiveProducers,
long upperBoundOffsetOfCleaningRound,
CleanerStats stats,
- long currentTime) throws IOException {
+ long currentTime,
+ int maxRecordBodySize) throws IOException {
MemoryRecords.RecordFilter logCleanerFilter = new MemoryRecords.RecordFilter(currentTime, deleteRetentionMs) {
private boolean discardBatchRecords;
@@ -388,7 +392,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());
@@ -669,7 +673,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",
@@ -691,6 +696,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
*/
@@ -701,7 +708,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);
@@ -729,7 +737,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 a687f3c529e..d4c2f55a766 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
@@ -26,6 +26,7 @@ import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.common.config.TopicConfig;
import org.apache.kafka.common.errors.InvalidConfigurationException;
import org.apache.kafka.common.record.CompressionType;
+import org.apache.kafka.common.record.Records;
import org.apache.kafka.common.record.TimestampType;
import org.apache.kafka.common.utils.ConfigUtils;
import org.apache.kafka.common.utils.Utils;
@@ -138,6 +139,10 @@ public class LogConfig extends AbstractConfig {
public static final boolean DEFAULT_REMOTE_LOG_DELETE_ON_DISABLE_CONFIG = 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.";
@@ -202,6 +207,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,
@@ -288,6 +295,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) {
@@ -310,6 +318,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);
@@ -365,6 +374,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 e33adb354dd..91bd43dc6c9 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
@@ -745,15 +745,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 a6137ffc681..2c77e4e8acf 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.MutableRecordBatch;
import org.apache.kafka.common.record.Record;
import org.apache.kafka.common.record.RecordBatch;
import org.apache.kafka.common.record.RecordValidationStats;
+import org.apache.kafka.common.record.Records;
import org.apache.kafka.common.record.TimestampType;
import org.apache.kafka.common.requests.ProduceResponse.RecordError;
import org.apache.kafka.common.utils.BufferSupplier;
@@ -79,6 +80,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,
@@ -92,6 +94,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;
@@ -104,6 +124,7 @@ public class LogValidator {
this.timestampAfterMaxMs = timestampAfterMaxMs;
this.partitionLeaderEpoch = partitionLeaderEpoch;
this.origin = origin;
+ this.maxRecordBodySize = maxRecordBodySize;
}
/**
@@ -313,9 +334,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 dc80383dc66..cf88c617fe0 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
@@ -1136,7 +1136,8 @@ public class UnifiedLog implements AutoCloseable {
config().messageTimestampBeforeMaxMs,
config().messageTimestampAfterMaxMs,
leaderEpoch,
- origin
+ origin,
+ config().maxDecompressedMessageBytes()
);
LogValidator.ValidationResult validateAndOffsetAssignResult = validator.validateMessagesAndAssignOffsets(offset,
validatorMetricsRecorder,
@@ -1721,7 +1722,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,
@@ -1783,7 +1784,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 5b0676088e1..ca244204169 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;
@@ -1653,6 +1655,7 @@ public class RemoteLogManagerTest {
.thenAnswer(a -> new ByteArrayInputStream(records(ts, startOffset, targetLeaderEpoch).buffer().array()));
when(mockLog.logEndOffset()).thenReturn(600L);
+ when(mockLog.config()).thenReturn(new LogConfig(new Properties()));
remoteLogManager.onLeadershipChange(Set.of(mockPartition(leaderTopicIdPartition)), Set.of(), topicIds);
}
@@ -1723,6 +1726,7 @@ public class RemoteLogManagerTest {
});
when(mockLog.logEndOffset()).thenReturn(300L);
+ when(mockLog.config()).thenReturn(new LogConfig(new Properties()));
remoteLogManager = new RemoteLogManager(config, brokerId, logDir, clusterId, time,
partition -> Optional.of(mockLog),
(topicPartition, offset) -> currentLogStartOffset.set(offset),
@@ -1732,7 +1736,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);
}
};
@@ -1762,7 +1766,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;
@@ -3868,4 +3872,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/LogSegmentTest.java b/storage/src/test/java/org/apache/kafka/storage/internals/log/LogSegmentTest.java
index 2131aaf7ed9..67dbfc117db 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
@@ -386,17 +386,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);
}
}
@@ -586,9 +586,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/LogValidatorTest.java b/storage/src/test/java/org/apache/kafka/storage/internals/log/LogValidatorTest.java
index 6ddf75e6e60..3910ed08d1b 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
@@ -189,6 +189,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) {