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 05:13:35 UTC
(kafka) branch 4.2 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.2
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/4.2 by this push:
new d67699ebaf1 MINOR: Add policy validation to CreatePartitions (#23669)
d67699ebaf1 is described below
commit d67699ebaf1f4e9e2fc30d74adf00a8e6e85f748
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>
Co-Authored-By: Claude Sonnet 5.5 <no...@anthropic.com>
---
.../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 8bd37ab5265..659145859e2 100644
--- a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
+++ b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
@@ -1222,6 +1222,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.
*
@@ -1811,6 +1841,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 7024d0de3a2..edc847b8dea 100644
--- a/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java
+++ b/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java
@@ -1786,6 +1786,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());
+ 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();