You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ma...@apache.org on 2026/09/25 05:32:48 UTC
(kafka) branch trunk updated: MINOR: Apply MessageUtil array limits in ArrayOf and CompactArrayOf
This is an automated email from the ASF dual-hosted git repository.
omkreddy pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new d89cb613d00 MINOR: Apply MessageUtil array limits in ArrayOf and CompactArrayOf
d89cb613d00 is described below
commit d89cb613d002957af869e083517b129f04dd7686
Author: Manikumar Reddy <ma...@confluent.io>
AuthorDate: Fri Sep 25 10:19:23 2026 +0530
MINOR: Apply MessageUtil array limits in ArrayOf and CompactArrayOf
Use the same array length limits as the generated message readers.
Co-authored-by: Anastasia Vela <av...@confluent.io>
Co-Authored-By: Claude Opus 5.5 <no...@anthropic.com>
---
.../kafka/common/protocol/types/ArrayOf.java | 12 ++++++++---
.../common/protocol/types/CompactArrayOf.java | 12 ++++++++---
.../protocol/types/ProtocolSerializationTest.java | 25 ++++++++++++++++++++++
3 files changed, 43 insertions(+), 6 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/types/ArrayOf.java b/clients/src/main/java/org/apache/kafka/common/protocol/types/ArrayOf.java
index 17c827744b3..fe6cfef7f78 100644
--- a/clients/src/main/java/org/apache/kafka/common/protocol/types/ArrayOf.java
+++ b/clients/src/main/java/org/apache/kafka/common/protocol/types/ArrayOf.java
@@ -16,9 +16,12 @@
*/
package org.apache.kafka.common.protocol.types;
+import org.apache.kafka.common.protocol.MessageUtil;
import org.apache.kafka.common.protocol.types.Type.DocumentedType;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Optional;
/**
@@ -76,10 +79,13 @@ public class ArrayOf extends DocumentedType {
if (size > buffer.remaining())
throw new SchemaException("Error reading array of size " + size + ", only " + buffer.remaining() + " bytes available");
- Object[] objs = new Object[size];
+ if (size > MessageUtil.MAX_ARRAY_LENGTH)
+ throw new SchemaException("Error reading array of size " + size + ", which exceeds the maximum allowed size of " + MessageUtil.MAX_ARRAY_LENGTH);
+
+ List<Object> values = new ArrayList<>(Math.min(size, MessageUtil.MAX_PREALLOCATED_ARRAY_CAPACITY));
for (int i = 0; i < size; i++)
- objs[i] = type.read(buffer);
- return objs;
+ values.add(type.read(buffer));
+ return values.toArray();
}
@Override
diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/types/CompactArrayOf.java b/clients/src/main/java/org/apache/kafka/common/protocol/types/CompactArrayOf.java
index 219858c28ed..0b2685fc639 100644
--- a/clients/src/main/java/org/apache/kafka/common/protocol/types/CompactArrayOf.java
+++ b/clients/src/main/java/org/apache/kafka/common/protocol/types/CompactArrayOf.java
@@ -16,10 +16,13 @@
*/
package org.apache.kafka.common.protocol.types;
+import org.apache.kafka.common.protocol.MessageUtil;
import org.apache.kafka.common.protocol.types.Type.DocumentedType;
import org.apache.kafka.common.utils.internals.ByteUtils;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Optional;
/**
@@ -81,10 +84,13 @@ public class CompactArrayOf extends DocumentedType {
int size = n - 1;
if (size > buffer.remaining())
throw new SchemaException("Error reading array of size " + size + ", only " + buffer.remaining() + " bytes available");
- Object[] objs = new Object[size];
+ if (size > MessageUtil.MAX_ARRAY_LENGTH)
+ throw new SchemaException("Error reading array of size " + size + ", which exceeds the maximum allowed size of " + MessageUtil.MAX_ARRAY_LENGTH);
+
+ List<Object> values = new ArrayList<>(Math.min(size, MessageUtil.MAX_PREALLOCATED_ARRAY_CAPACITY));
for (int i = 0; i < size; i++)
- objs[i] = type.read(buffer);
- return objs;
+ values.add(type.read(buffer));
+ return values.toArray();
}
@Override
diff --git a/clients/src/test/java/org/apache/kafka/common/protocol/types/ProtocolSerializationTest.java b/clients/src/test/java/org/apache/kafka/common/protocol/types/ProtocolSerializationTest.java
index fd942f218b3..4616422bb71 100644
--- a/clients/src/test/java/org/apache/kafka/common/protocol/types/ProtocolSerializationTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/protocol/types/ProtocolSerializationTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.common.protocol.types;
+import org.apache.kafka.common.protocol.MessageUtil;
import org.apache.kafka.common.utils.internals.ByteUtils;
import org.junit.jupiter.api.BeforeEach;
@@ -221,6 +222,30 @@ public class ProtocolSerializationTest {
"Array size not validated");
}
+ @Test
+ public void testReadArrayLengthAboveMaxIsRejected() {
+ Type type = new ArrayOf(Type.INT8);
+ int count = MessageUtil.MAX_ARRAY_LENGTH + 1;
+ ByteBuffer buffer = ByteBuffer.allocate(4 + count);
+ buffer.putInt(count);
+ buffer.rewind();
+ SchemaException e = assertThrows(SchemaException.class, () -> type.read(buffer));
+ assertEquals("Error reading array of size " + count + ", which exceeds the maximum allowed size of "
+ + MessageUtil.MAX_ARRAY_LENGTH, e.getMessage());
+ }
+
+ @Test
+ public void testReadCompactArrayLengthAboveMaxIsRejected() {
+ Type type = new CompactArrayOf(Type.INT8);
+ int count = MessageUtil.MAX_ARRAY_LENGTH + 1;
+ ByteBuffer buffer = ByteBuffer.allocate(ByteUtils.sizeOfUnsignedVarint(count + 1) + count);
+ ByteUtils.writeUnsignedVarint(count + 1, buffer);
+ buffer.rewind();
+ SchemaException e = assertThrows(SchemaException.class, () -> type.read(buffer));
+ assertEquals("Error reading array of size " + count + ", which exceeds the maximum allowed size of "
+ + MessageUtil.MAX_ARRAY_LENGTH, e.getMessage());
+ }
+
@Test
public void testReadTaggedFieldsSizeTooLarge() {
int tag = 1;