(kafka) branch trunk updated: MINOR: Prevent overflow in CreateTopics partition validation (#22722)

[mimaison]

This is an automated email from the ASF dual-hosted git repository.

mimaison 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 e74637ae041 MINOR: Prevent overflow in CreateTopics partition 
validation (#22722)
e74637ae041 is described below

commit e74637ae04152a720c9e28937700e222dfa53e91
Author: CoderBruis <37364336+coderbr...@users.noreply.github.com>
AuthorDate: Wed Jul 8 21:28:21 2026 +0800

    MINOR: Prevent overflow in CreateTopics partition validation (#22722)
    
    This patch updates the CreateTopics partition count validation.
    
    Changes:
    
    * Update the validation comment to state that a request may create at
    most `MAX_PARTITIONS_PER_BATCH` partitions.
    * Use a `long` counter to prevent integer overflow when accumulating
    partition counts from the request.
    * Move the limit check inside the loop so that oversized requests fail
    as soon as the limit is exceeded.
    
    Test:
    
    * Add `testTotalNumberOfPartitionsValidationDoesNotOverflow` to cover a
    CreateTopics request large enough to overflow an `int`.
    
    Reviewers: Mickael Maison <mickael.mai...@gmail.com>
---
 .../kafka/controller/ReplicationControlManager.java       | 11 +++++------
 .../kafka/controller/ReplicationControlManagerTest.java   | 15 +++++++++++++++
 2 files changed, 20 insertions(+), 6 deletions(-)

diff --git 
a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
 
b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
index ae1854f3ced..4c8a24f0faa 100644
--- 
a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
+++ 
b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
@@ -1211,7 +1211,7 @@ public class ReplicationControlManager {
     }
 
     /**
-     * Validates that a batch of topics will create less than {@value 
MAX_PARTITIONS_PER_BATCH}. Exceeding this number of topics per batch
+     * Validates that a batch of topics will create at most {@value 
MAX_PARTITIONS_PER_BATCH}. Exceeding this number of topics per batch
      * has led to out-of-memory exceptions. We use this validation to fail 
earlier to avoid allocating the memory.
      * Validates an upper bound number of partitions. The actual number may be 
smaller if some topics are misconfigured.
      *
@@ -1220,7 +1220,7 @@ public class ReplicationControlManager {
      * @throws PolicyViolationException if total number of partitions exceeds 
{@value MAX_PARTITIONS_PER_BATCH}.
      */
     static void validateTotalNumberOfPartitions(CreateTopicsRequestData 
request, int defaultNumPartitions) {
-        int totalPartitions = 0;
+        long totalPartitions = 0;
         for (CreatableTopic topic: request.topics()) {
             if (topic.assignments().isEmpty()) {
                 if (topic.numPartitions() == -1) {
@@ -1231,10 +1231,9 @@ public class ReplicationControlManager {
             } else {
                 totalPartitions += topic.assignments().size();
             }
-
-        }
-        if (totalPartitions > MAX_PARTITIONS_PER_BATCH) {
-            throw new PolicyViolationException("Excessively large number of 
partitions per request.");
+            if (totalPartitions > MAX_PARTITIONS_PER_BATCH) {
+                throw new PolicyViolationException("Excessively large number 
of partitions per request.");
+            }
         }
     }
 
diff --git 
a/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java
 
b/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java
index cb7a74b334c..b6b66a906e8 100644
--- 
a/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java
+++ 
b/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java
@@ -621,6 +621,21 @@ public class ReplicationControlManagerTest {
         assertEquals(error.getMessage(), "Excessively large number of 
partitions per request.");
     }
 
+    @Test
+    public void testTotalNumberOfPartitionsValidationDoesNotOverflow() {
+        CreateTopicsRequestData request = new CreateTopicsRequestData();
+        request.topics().add(new CreatableTopic().setName("foo").
+                
setNumPartitions(Integer.MAX_VALUE).setReplicationFactor((short) 1));
+        request.topics().add(new CreatableTopic().setName("bar").
+                
setNumPartitions(Integer.MAX_VALUE).setReplicationFactor((short) 1));
+
+        PolicyViolationException error = assertThrows(
+                PolicyViolationException.class,
+                () -> 
ReplicationControlManager.validateTotalNumberOfPartitions(request, 1)
+        );
+        assertEquals("Excessively large number of partitions per request.", 
error.getMessage());
+    }
+
     @Test
     public void testCreateTopics() {
         ReplicationControlTestContext ctx = new 
ReplicationControlTestContext.Builder().build();