This is an automated email from the ASF dual-hosted git repository. lhotari pushed a commit to branch branch-3.0 in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 386f6f0bf73d01f7b23160e9e4642f4f7c01cac5 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 c674a035653..ae6928d2b32 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 @@ -1348,7 +1348,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 602cd47e595..dac3e7cd55d 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 @@ -433,7 +433,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("Expect partitions %s grater than maximum partitions per topic %s", expectPartitions, brokerMaximumPartitionsPerTopic)); 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 d7ffa656bdb..0eddbf1fea1 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 @@ -1658,6 +1658,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