diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml index ff60f0d0fbb78..703fea90c5d55 100644 --- a/checkstyle/suppressions.xml +++ b/checkstyle/suppressions.xml @@ -362,9 +362,11 @@ files="(LocalLog|LogLoader|LogValidator|RemoteLogManager|RemoteIndexCache|UnifiedLog).java"/> + - + 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 e97c39bc61911..8d18b666f14eb 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 " + + "InvalidRecordException. 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 max.message.bytes, 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 e47d7c866ccc3..5d1f024811e36 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 Iterator iterator() { } CloseableIterator iterator(BufferSupplier bufferSupplier) { + return iterator(bufferSupplier, Records.SOFT_MAX_ARRAY_LENGTH); + } + + CloseableIterator 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 CloseableIterator streamingIterator(BufferSupplier bufferSupplier return iterator(bufferSupplier); } + @Override + public CloseableIterator 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 @@ static void writeHeader(DataOutputStream out, long offset, int size) throws IOEx private static final class DataLogInputStream implements LogInputStream { 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 AbstractLegacyRecordBatch nextBatch() throws IOException { 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 @@ private static class DeepRecordsIterator extends AbstractIterator implem 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 @@ private DeepRecordsIterator(AbstractLegacyRecordBatch wrapperEntry, wrapperMagic + ")"); InputStream stream = Compression.of(compressionType).build().wrapForInput(wrapperValue, wrapperRecord.magic(), bufferSupplier); - LogInputStream logStream = new DataLogInputStream(stream, maxMessageSize); + LogInputStream logStream = new DataLogInputStream(stream, maxMessageSize, maxRecordBodySize); long lastOffsetFromWrapper = wrapperEntry.lastOffset(); long timestampFromWrapper = wrapperRecord.timestamp(); @@ -517,6 +539,12 @@ public CloseableIterator skipKeyValueIterator(BufferSupplier bufferSuppl return CloseableIterator.wrap(iterator(bufferSupplier)); } + @Override + public CloseableIterator 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 783155a00619c..9a302cc0bc3d3 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 int hashCode() { 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 @@ private static DefaultRecord readFrom(ByteBuffer buffer, } } + /** + * 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 d6e9cc6bd7fbb..9c923938095e4 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 InputStream recordInputStream(BufferSupplier bufferSupplier) { return Compression.of(compressionType()).build().wrapForInput(buffer, magic(), bufferSupplier); } - private CloseableIterator compressedIterator(BufferSupplier bufferSupplier, boolean skipKeyValue) { + private CloseableIterator 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 Iterator iterator() { // 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 iterator = compressedIterator(BufferSupplier.NO_CACHING, false)) { + try (CloseableIterator iterator = compressedIterator(BufferSupplier.NO_CACHING, false, Records.SOFT_MAX_ARRAY_LENGTH)) { List records = new ArrayList<>(count()); while (iterator.hasNext()) records.add(iterator.next()); @@ -337,6 +337,11 @@ public Iterator iterator() { @Override public CloseableIterator skipKeyValueIterator(BufferSupplier bufferSupplier) { + return skipKeyValueIterator(bufferSupplier, Records.SOFT_MAX_ARRAY_LENGTH); + } + + @Override + public CloseableIterator skipKeyValueIterator(BufferSupplier bufferSupplier, int maxRecordBodySize) { if (count() == 0) { return CloseableIterator.wrap(Collections.emptyIterator()); } @@ -351,13 +356,18 @@ public CloseableIterator skipKeyValueIterator(BufferSupplier bufferSuppl // 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 streamingIterator(BufferSupplier bufferSupplier) { + return streamingIterator(bufferSupplier, Records.SOFT_MAX_ARRAY_LENGTH); + } + + @Override + public CloseableIterator 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 bd966e79f032d..358a9761b59a9 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 CloseableIterator streamingIterator(BufferSupplier bufferSupplier return loadFullBatch().streamingIterator(bufferSupplier); } + @Override + public CloseableIterator 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 2f5e2e50dde75..795a1f53e651d 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.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 LogOffsetPosition searchForOffsetFromPosition(long targetOffset, int star * @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 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 f806cce692073..32b3a3e700ae8 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 Integer firstBatchSize() { /** * 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 Integer firstBatchSize() { * 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 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 @@ private static FilterResult filterTo(Iterable batches, Recor continue; final BatchFilterResult iterationResult = filterBatch(batch, decompressionBufferSupplier, filterResult, - filter); + filter, maxRecordBodySize); List retainedRecords = iterationResult.retainedRecords; boolean containsTombstones = iterationResult.containsTombstones; boolean writeOriginalBatch = iterationResult.writeOriginalBatch; @@ -215,8 +214,9 @@ private static FilterResult filterTo(Iterable batches, Recor private static BatchFilterResult filterBatch(RecordBatch batch, BufferSupplier decompressionBufferSupplier, FilterResult filterResult, - RecordFilter filter) { - try (final CloseableIterator iterator = batch.streamingIterator(decompressionBufferSupplier)) { + RecordFilter filter, + int maxRecordBodySize) { + try (final CloseableIterator 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 b5f42e5b915fa..ba4e6aa19af64 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 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 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 e36beff08f2ac..87b0e13f0635e 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 { */ CloseableIterator 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 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 { * 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 offsetOfMaxTimestamp() { + default Optional offsetOfMaxTimestamp(int maxRecordBodySize) { if (magic() == RecordBatch.MAGIC_VALUE_V0) return Optional.empty(); long maxTimestamp = maxTimestamp(); - try (CloseableIterator iter = streamingIterator(BufferSupplier.create())) { + try (CloseableIterator 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 017c49ba94cdb..b6e2d294a5e6a 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 f8e53148b0100..d6cbbcb2f6822 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 @@ protected BatchRetentionResult checkBatchRetention(RecordBatch batch) { 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 93da0a4433af0..4e1530fb3c1e6 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 @@ protected BatchRetentionResult checkBatchRetention(RecordBatch batch) { 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 b6040950ccdd6..797074fcc04be 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.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 @@ protected BatchRetentionResult checkBatchRetention(RecordBatch batch) { 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 015cebec1b095..f741c6f5b8d8b 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 @@ 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 void testZStdCompressionTypeWithV0OrV1() { } } + /** + * 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 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 e63dafc3a0d01..b193e40598836 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.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.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 void testDecrementSequence() { 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 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 iterator = batch.streamingIterator(BufferSupplier.NO_CACHING, 10_000)) { + assertNotNull(iterator.next()); + } + try (CloseableIterator 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 iterator = batch.skipKeyValueIterator(BufferSupplier.NO_CACHING, 10_000)) { + assertNotNull(iterator.next()); + } + try (CloseableIterator 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 2473cca54109b..608d95f030fa8 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.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 void testBasicSerdeInvalidHeaderCountTooHigh() throws IOException { // 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 void testSerdeNoSequence() throws IOException { // 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 void testInvalidSizeOfBodyInBytes() throws IOException { 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 @@ private static void assertDecodingRecordFromBufferThrowsInvalidRecordException(B // 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 a9d12285f1298..cb91b90f43702 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 @@ private void testSearchForTimestamp(RecordVersion version) throws IOException { 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 @@ private void append(FileRecords fileRecords, byte[][] values) throws IOException } 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 7092928010b30..2b573e9788cd7 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 void testFilterToPreservesPartitionLeaderEpoch(Args args) { 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 @@ protected boolean shouldRetainRecord(RecordBatch recordBatch, Record record) { // 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 @@ protected BatchRetentionResult checkBatchRetention(RecordBatch batch) { 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 @@ protected BatchRetentionResult checkBatchRetention(RecordBatch batch) { 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 @@ protected BatchRetentionResult checkBatchRetention(RecordBatch batch) { 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 @@ protected BatchRetentionResult checkBatchRetention(RecordBatch batch) { 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 void testFilterToAlreadyCompactedLog(Args args) { 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 void testFilterToPreservesProducerInfo(Args args) { 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 void testFilterToWithUndersizedBuffer(Args args) { 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 void testFilterTo(Args args) { 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 void testFilterToPreservesLogAppendTime(Args args) { 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 743937b54fca5..a26d7281b631d 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 961ec31acbf50..8d32711bba30f 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 318a0c6614030..7905ded58b948 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 98a49070913b1..4ccb6c096621d 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 1881b3629ab66..acfd7ae8b6177 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 0155c1008c667..ab211be5fbe38 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 a20139cf23c49..14ba35786374d 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 99565e90084cd..65bbebf3d8120 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 57545c7ba2b00..3a6573e4123a4 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 e40c5c8e6d36f..4dfebcd4d89d8 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.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.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 " + TopicConfig.MAX_MESSAGE_BYTES_CONFIG + " 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 " + TopicConfig.MAX_DECOMPRESSED_MESSAGE_BYTES_CONFIG + " 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 5438a0a59cb6f..37db03f185eb4 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 c05f9f2816ae0..1145ad15b4cee 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 b104de4c15660..9ab6acb5b7bbd 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 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 static DescribeConfigsResponseData.DescribeConfigsResult createRespon } /** - * 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 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 explicitlySetConfigs, + BiFunction createConfigEntry) { // Cast from Map to Map to eliminate wildcard types. Cached to avoid multiple calls. @SuppressWarnings("unchecked") Map nonInternalValues = (Map) config.nonInternalValues(); Stream> 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 a9848633effa3..d1d90a605ebb0 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 Optional fetchNextSegmentWithTxnIndex(TopicPart return remoteLogMetadataManagerPlugin.get().nextSegmentWithTxnIndex(tpId, epochForOffset, offset); } - Optional lookupTimestamp(RemoteLogSegmentMetadata rlsMetadata, long timestamp, long startingOffset) - throws RemoteStorageException, IOException { + Optional 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 @@ Optional lookupTimestamp(RemoteLogSegmentMetadat RecordBatch batch = remoteLogInputStream.nextBatch(); if (batch == null) break; if (batch.maxTimestamp() >= timestamp && batch.lastOffset() >= startingOffset) { - try (CloseableIterator recordStreamingIterator = batch.streamingIterator(BufferSupplier.NO_CACHING)) { + try (CloseableIterator 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 Optional findOffsetByTimestamp(TopicParti 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 @@ && isRemoteSegmentWithinLeaderEpochs(rlsMetadata, unifiedLog.logEndOffset(), epo List 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 e17809d988ece..00a2345df1dc6 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 void cleanSegments(UnifiedLog log, 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 void cleanSegments(UnifiedLog log, * @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 @@ private void cleanInto(TopicPartition topicPartition, Map 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 boolean shouldRetainRecord(RecordBatch batch, Record record) { 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 void buildOffsetMap(UnifiedLog log, 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 void buildOffsetMap(UnifiedLog log, * @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 @@ private boolean buildOffsetMapForSegment(TopicPartition topicPartition, 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 @@ private boolean buildOffsetMapForSegment(TopicPartition topicPartition, // Note that abort markers are supported in v2 and above, which means count is defined. stats.indexMessagesRead(batch.countOrNull()); } else { - try (CloseableIterator recordsIterator = batch.streamingIterator(decompressionBufferSupplier)) { + try (CloseableIterator recordsIterator = batch.streamingIterator(decompressionBufferSupplier, maxRecordBodySize)) { for (Record record : (Iterable) () -> 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 a687f3c529e32..d4c2f55a766c7 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.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 Optional serverConfigName(String configName) { 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 Optional serverConfigName(String configName) { 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 Optional serverConfigName(String configName) { 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 LogConfig(Map props, Set overriddenConfigs) { 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 int maxMessageSize() { 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 e33adb354dd02..91bd43dc6c9b0 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 long getFirstBatchTimestamp() { * * @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 findOffsetByTimestamp(long timestampMs, long startingOffset) throws IOException { + public Optional 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 a6137ffc68131..2c77e4e8acfc8 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.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 @@ private record ApiRecordError(Errors apiError, RecordError recordError) { 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 LogValidator(MemoryRecords records, 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 LogValidator(MemoryRecords records, this.timestampAfterMaxMs = timestampAfterMaxMs; this.partitionLeaderEpoch = partitionLeaderEpoch; this.origin = origin; + this.maxRecordBodySize = maxRecordBodySize; } /** @@ -313,9 +334,9 @@ public ValidationResult validateMessagesAndAssignOffsetsCompressed(LongRef offse // then we can optimize the iterator to skip key / value / headers since they would not be used at all CloseableIterator 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 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 dc80383dc668b..cf88c617fe074 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 @@ private LogAppendInfo append(MemoryRecords records, config().messageTimestampBeforeMaxMs, config().messageTimestampAfterMaxMs, leaderEpoch, - origin + origin, + config().maxDecompressedMessageBytes() ); LogValidator.ValidationResult validateAndOffsetAssignResult = validator.validateMessagesAndAssignOffsets(offset, validatorMetricsRecorder, @@ -1721,7 +1722,7 @@ public OffsetResultHolder fetchOffsetByTimestamp(long targetTimestamp, Optional< Optional 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 @@ private Optional searchOffsetInLocalLog(long tar List 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 5b0676088e1e5..ca244204169f1 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 @@ private void doTestFindOffsetByTimestamp(long ts, long startOffset, int targetLe .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 @@ void testFetchOffsetByTimestampWithTieredStorageDoesNotFetchIndexWhenExistsLocal }); 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 RemoteLogMetadataManager createRemoteLogMetadataManager() { return remoteLogMetadataManager; } @Override - Optional lookupTimestamp(RemoteLogSegmentMetadata rlsMetadata, long timestamp, long startingOffset) { + Optional lookupTimestamp(RemoteLogSegmentMetadata rlsMetadata, long timestamp, long startingOffset, int maxRecordBodySize) { return Optional.of(expectedRemoteResult); } }; @@ -1762,7 +1766,7 @@ private LogSegment mockLogSegment(long baseOffset, 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 void withPluginMetrics(PluginMetrics metrics) { pluginMetrics = true; } } + + + @Test + void testFindOffsetByTimestampRejectsRemoteRecordExceedingMaxDecompressedMessageBytes() throws IOException, RemoteStorageException { + TopicPartition tp = leaderTopicIdPartition.topicPartition(); + long ts = time.milliseconds(); + long startOffset = 120; + int targetLeaderEpoch = 10; + + TreeMap 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 2131aaf7ed98b..67dbfc117db19 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 void testFindOffsetByTimestamp() throws IOException { 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 void testRecoveryFixesCorruptTimeIndex() throws IOException { 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 6ddf75e6e60a7..3910ed08d1bc4 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 void testBatchWithoutRecordsNotAllowed(String sourceCompressionName, Stri )); } + /** + * 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) {