This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 49bcbb6c3da2246bd887605fd68c10e1d3e9af28
Author: hanmz <gunn...@tencent.com>
AuthorDate: Wed Apr 10 04:27:22 2024 +0800
[fix][broker] Update topic partition failed when config
maxNumPartitionsPerPartitionedTopic<0 (#22397)
(cherry picked from commit fb5caeb2cd3353db0499e32e9ec79390741b809c)
---
.../apache/pulsar/broker/ServiceConfiguration.java | 3 +-
.../broker/admin/impl/PersistentTopicsBase.java | 2 +-
.../pulsar/broker/admin/PersistentTopicsTest.java | 45 ++++++++++++++++++++++
3 files changed, 48 insertions(+), 2 deletions(-)
diff --git
a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
index 8b99501f296..2e7d4d582d6 100644
---
a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
+++
b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -1351,7 +1351,8 @@ public class ServiceConfiguration implements
PulsarConfiguration {
category = CATEGORY_SERVER,
dynamic = true,
doc = "The number of partitions per partitioned topic.\n"
- + "If try to create or update partitioned topics by exceeded
number of partitions, then fail."
+ + "If try to create or update partitioned topics by exceeded
number of partitions, then fail.\n"
+ + "Use 0 or negative number to disable the check."
)
private int maxNumPartitionsPerPartitionedTopic = 0;
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
index 318e2bc2cde..b0968f494ee 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
@@ -348,7 +348,7 @@ public class PersistentTopicsBase extends AdminResource {
}
int brokerMaximumPartitionsPerTopic =
pulsarService.getConfiguration()
.getMaxNumPartitionsPerPartitionedTopic();
- if (brokerMaximumPartitionsPerTopic != 0 && expectPartitions >
brokerMaximumPartitionsPerTopic) {
+ if (brokerMaximumPartitionsPerTopic > 0 && expectPartitions >
brokerMaximumPartitionsPerTopic) {
throw new RestException(422 /* Unprocessable entity*/,
String.format("Desired partitions %s can't be
greater than the maximum partitions per"
+ " topic %s.",
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
index 8e1375303ce..c588051a0fe 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
@@ -1742,6 +1742,51 @@ public class PersistentTopicsTest extends
MockedPulsarServiceBaseTest {
verify(response, timeout(5000).times(1)).resume(metaCaptor.capture());
partitionedTopicMetadata = metaCaptor.getValue();
Assert.assertEquals(partitionedTopicMetadata.partitions, 4);
+
+ // test for configuration maxNumPartitionsPerPartitionedTopic
+ conf.setMaxNumPartitionsPerPartitionedTopic(4);
+ response = mock(AsyncResponse.class);
+ throwableCaptor = ArgumentCaptor.forClass(Throwable.class);
+ persistentTopics.updatePartitionedTopic(response, testTenant,
testNamespaceLocal, topicName, false, true,
+ true, 5);
+ verify(response,
timeout(5000).times(1)).resume(throwableCaptor.capture());
+ Assert.assertEquals(throwableCaptor.getValue().getMessage(),
+ "Desired partitions 5 can't be greater than the maximum
partitions per topic 4.");
+
+ response = mock(AsyncResponse.class);
+ metaCaptor = ArgumentCaptor.forClass(PartitionedTopicMetadata.class);
+ persistentTopics.getPartitionedMetadata(response, testTenant,
testNamespaceLocal, topicName, true, false);
+ verify(response, timeout(5000).times(1)).resume(metaCaptor.capture());
+ partitionedTopicMetadata = metaCaptor.getValue();
+ Assert.assertEquals(partitionedTopicMetadata.partitions, 4);
+
+ conf.setMaxNumPartitionsPerPartitionedTopic(-1);
+ response = mock(AsyncResponse.class);
+ responseCaptor = ArgumentCaptor.forClass(Response.class);
+ persistentTopics.updatePartitionedTopic(response, testTenant,
testNamespaceLocal, topicName, false, true,
+ true, 5);
+ verify(response,
timeout(5000).times(1)).resume(responseCaptor.capture());
+
+ response = mock(AsyncResponse.class);
+ metaCaptor = ArgumentCaptor.forClass(PartitionedTopicMetadata.class);
+ persistentTopics.getPartitionedMetadata(response, testTenant,
testNamespaceLocal, topicName, true, false);
+ verify(response, timeout(5000).times(1)).resume(metaCaptor.capture());
+ partitionedTopicMetadata = metaCaptor.getValue();
+ Assert.assertEquals(partitionedTopicMetadata.partitions, 5);
+
+ conf.setMaxNumPartitionsPerPartitionedTopic(0);
+ response = mock(AsyncResponse.class);
+ responseCaptor = ArgumentCaptor.forClass(Response.class);
+ persistentTopics.updatePartitionedTopic(response, testTenant,
testNamespaceLocal, topicName, false, true,
+ true, 6);
+ verify(response,
timeout(5000).times(1)).resume(responseCaptor.capture());
+
+ response = mock(AsyncResponse.class);
+ metaCaptor = ArgumentCaptor.forClass(PartitionedTopicMetadata.class);
+ persistentTopics.getPartitionedMetadata(response, testTenant,
testNamespaceLocal, topicName, true, false);
+ verify(response, timeout(5000).times(1)).resume(metaCaptor.capture());
+ partitionedTopicMetadata = metaCaptor.getValue();
+ Assert.assertEquals(partitionedTopicMetadata.partitions, 6);
}
@Test
