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/10/02 04:52:46 UTC
(kafka) branch 4.4 updated: MINOR: Add policy validation to CreatePartitions (#23669)
This is an automated email from the ASF dual-hosted git repository.
omkreddy pushed a commit to branch 4.4
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/4.4 by this push:
new 16ab9e3775d MINOR: Add policy validation to CreatePartitions (#23669)
16ab9e3775d is described below
commit 16ab9e3775dbbbc9b35aa84c097df24302524a2a
Author: Alyssa Huang <ah...@confluent.io>
AuthorDate: Thu Oct 1 21:52:37 2026 -0700
MINOR: Add policy validation to CreatePartitions (#23669)
Add same CreateTopics validation to CreatePartitions
Reviewers: Manikumar Reddy <ma...@confluent.io>
---
.../controller/ReplicationControlManager.java | 31 ++++
.../controller/ReplicationControlManagerTest.java | 159 +++++++++++++++++++++
2 files changed, 190 insertions(+)
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 c915615e1ff..a3c03c0b135 100644
--- a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
+++ b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
@@ -1237,6 +1237,36 @@ public class ReplicationControlManager {
}
}
+ /**
+ * Validates that a batch of partition changes will create at most {@value MAX_PARTITIONS_PER_BATCH} additional partitions.
+ *
+ * @param topicsToCreate a batch of new partitions to create.
+ *
+ * @throws PolicyViolationException if total number of partitions exceeds {@value MAX_PARTITIONS_PER_BATCH}.
+ */
+ void validateTotalNumberOfPartitions(List<CreatePartitionsTopic> topicsToCreate) {
+ long totalAdditionalPartitions = 0;
+ for (CreatePartitionsTopic topic : topicsToCreate) {
+ if (topic.count() < 0) {
+ continue;
+ }
+ Uuid topicId = this.topicsByName.get(topic.name());
+ if (topicId == null) {
+ continue;
+ }
+ int additionalPartitions = topic.count() - topics.get(topicId).parts.size();
+ if (additionalPartitions <= 0) {
+ continue;
+ }
+ totalAdditionalPartitions += additionalPartitions;
+ if (totalAdditionalPartitions > MAX_PARTITIONS_PER_BATCH) {
+ throw new PolicyViolationException(
+ "Excessively large number of additional partitions per request."
+ );
+ }
+ }
+ }
+
/**
* Validate the partition information included in the alter partition request.
*
@@ -1882,6 +1912,7 @@ public class ReplicationControlManager {
ControllerRequestContext context,
List<CreatePartitionsTopic> topics
) {
+ validateTotalNumberOfPartitions(topics);
List<ApiMessageAndVersion> records = BoundedList.newArrayBacked(MAX_RECORDS_PER_USER_OP);
List<CreatePartitionsTopicResult> results = BoundedList.newArrayBacked(MAX_RECORDS_PER_USER_OP);
for (CreatePartitionsTopic topic : topics) {
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 71dc3c00cfe..dd1e636c38b 100644
--- a/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java
+++ b/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java
@@ -1810,6 +1810,165 @@ public class ReplicationControlManagerTest {
((TopicRecord) result.records().get(0).message()).topicId(), 1));
}
+ @Test
+ public void testCreatePartitionsWithOverSizedBatch() {
+ ReplicationControlTestContext ctx = new ReplicationControlTestContext.Builder().build();
+ ReplicationControlManager replicationControl = ctx.replicationControl;
+ CreateTopicsRequestData request = new CreateTopicsRequestData();
+ int initialPartitions = 2;
+ request.topics().add(
+ new CreatableTopic().
+ setName("foo").
+ setNumPartitions(initialPartitions).
+ setReplicationFactor((short) 2)
+ );
+ request.topics().add(
+ new CreatableTopic().
+ setName("bar").
+ setNumPartitions(initialPartitions).
+ setReplicationFactor((short) 2)
+ );
+ request.topics().add(
+ new CreatableTopic().
+ setName("baz").
+ setNumPartitions(initialPartitions).
+ setReplicationFactor((short) 2)
+ );
+ ctx.registerBrokers(0, 1, 2);
+ ctx.unfenceBrokers(0, 1, 2);
+ ControllerResult<CreateTopicsResponseData> createTopicResult = replicationControl.
+ createTopics(anonymousContextFor(ApiKeys.CREATE_TOPICS), request, Set.of(), false);
+ ctx.replay(createTopicResult.records());
+
+ ControllerRequestContext requestContext = anonymousContextFor(ApiKeys.CREATE_PARTITIONS);
+
+ // A single extremely large, existing topic should trip the error.
+ List<CreatePartitionsTopic> oneLargePartition = List.of(
+ new CreatePartitionsTopic().
+ setName("foo").
+ setCount(Integer.MAX_VALUE).
+ setAssignments(null)
+ );
+ Throwable t = assertThrows(
+ PolicyViolationException.class,
+ () -> replicationControl.createPartitions(requestContext, oneLargePartition)
+ );
+ assertEquals(
+ "Excessively large number of additional partitions per request.",
+ t.getMessage()
+ );
+
+ // Negative counts are ignored.
+ List<CreatePartitionsTopic> negativeOnly = List.of(
+ new CreatePartitionsTopic().
+ setName("foo").
+ setCount(Integer.MIN_VALUE).
+ setAssignments(null)
+ );
+ ControllerResult<List<CreatePartitionsTopicResult>> r1 =
+ replicationControl.createPartitions(requestContext, negativeOnly);
+ assertEquals(
+ List.of(
+ new CreatePartitionsTopicResult().
+ setName("foo").
+ setErrorCode(INVALID_PARTITIONS.code()).
+ setErrorMessage("The topic foo currently has 2 partition(s); -2147483648 would not be an increase.")
+ ),
+ r1.response()
+ );
+
+ // Negative and positive counts do not sum together.
+ List<CreatePartitionsTopic> positiveAndNegative = List.of(
+ new CreatePartitionsTopic().
+ setName("foo").
+ setCount(3).
+ setAssignments(null),
+ new CreatePartitionsTopic().
+ setName("bar").
+ setCount(Integer.MIN_VALUE).
+ setAssignments(null),
+ new CreatePartitionsTopic().
+ setName("baz").
+ setCount(Integer.MAX_VALUE).
+ setAssignments(null)
+ );
+ assertThrows(
+ PolicyViolationException.class,
+ () -> replicationControl.createPartitions(requestContext, positiveAndNegative)
+ );
+
+ // Do not throw if additional partitions exactly equals ReplicationControlManager.MAX_PARTITIONS_PER_BATCH
+ List<CreatePartitionsTopic> exactlyAtLimit = List.of(
+ new CreatePartitionsTopic().
+ setName("foo").
+ setCount(initialPartitions + 1).
+ setAssignments(null),
+ new CreatePartitionsTopic().
+ setName("bar").
+ setCount(ReplicationControlManager.MAX_PARTITIONS_PER_BATCH / 2 + initialPartitions - 1).
+ setAssignments(null),
+ new CreatePartitionsTopic().
+ setName("baz").
+ setCount(ReplicationControlManager.MAX_PARTITIONS_PER_BATCH / 2 + initialPartitions).
+ setAssignments(null)
+ );
+ ControllerResult<List<CreatePartitionsTopicResult>> r2 =
+ replicationControl.createPartitions(requestContext, exactlyAtLimit);
+ assertEquals(
+ List.of(
+ new CreatePartitionsTopicResult().
+ setName("foo").
+ setErrorCode(NONE.code()),
+ new CreatePartitionsTopicResult().
+ setName("bar").
+ setErrorCode(NONE.code()),
+ new CreatePartitionsTopicResult().
+ setName("baz").
+ setErrorCode(NONE.code())
+ ),
+ r2.response()
+ );
+
+ // Topics which do not exist do not count towards the limit.
+ // Topics which attempt to reduce the number of partitions do not count towards the total.
+ List<CreatePartitionsTopic> errorResultsDontCount = List.of(
+ new CreatePartitionsTopic().
+ setName("foo").
+ setCount(initialPartitions + ReplicationControlManager.MAX_PARTITIONS_PER_BATCH).
+ setAssignments(null),
+ new CreatePartitionsTopic().
+ setName("quux").
+ setCount(Integer.MAX_VALUE).
+ setAssignments(null),
+ new CreatePartitionsTopic().
+ setName("bar").
+ setCount(initialPartitions - 1).
+ setAssignments(
+ List.of(
+ new CreatePartitionsAssignment().setBrokerIds(List.of(1, 2))
+ )
+ )
+ );
+ ControllerResult<List<CreatePartitionsTopicResult>> r3 =
+ replicationControl.createPartitions(requestContext, errorResultsDontCount);
+ assertEquals(
+ List.of(
+ new CreatePartitionsTopicResult().
+ setName("foo").
+ setErrorCode(NONE.code()),
+ new CreatePartitionsTopicResult().
+ setName("quux").
+ setErrorCode(UNKNOWN_TOPIC_OR_PARTITION.code()).
+ setErrorMessage(null),
+ new CreatePartitionsTopicResult().
+ setName("bar").
+ setErrorCode(INVALID_PARTITIONS.code()).
+ setErrorMessage("The topic bar currently has 2 partition(s); 1 would not be an increase.")
+ ),
+ r3.response()
+ );
+ }
+
@Test
public void testValidateGoodManualPartitionAssignments() {
ReplicationControlTestContext ctx = new ReplicationControlTestContext.Builder().build();