This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.10
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.10 by this push:
new 56ef04c3a0c [fix][broker] Fix NPE when set `AutoTopicCreationOverride`
(#15653)
56ef04c3a0c is described below
commit 56ef04c3a0c2398311ff292604428921c051182b
Author: Qiang Zhao <[email protected]>
AuthorDate: Fri May 20 15:21:47 2022 +0800
[fix][broker] Fix NPE when set `AutoTopicCreationOverride` (#15653)
(cherry picked from commit e2afcf0a7ee90908fa647656624aacb9fd93249c)
---
.../pulsar/broker/admin/impl/NamespacesBase.java | 9 +++--
.../impl/ModularLoadManagerWrapper.java | 5 ++-
.../apache/pulsar/broker/admin/AdminApi2Test.java | 44 ++++++++++++++++++++++
3 files changed, 53 insertions(+), 5 deletions(-)
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
index a3ae333fd77..4f589a5684e 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
@@ -94,6 +94,7 @@ import
org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
import org.apache.pulsar.common.policies.data.SubscribeRate;
import org.apache.pulsar.common.policies.data.SubscriptionAuthMode;
import org.apache.pulsar.common.policies.data.TenantOperation;
+import org.apache.pulsar.common.policies.data.TopicType;
import org.apache.pulsar.common.policies.data.ValidateResult;
import
org.apache.pulsar.common.policies.data.impl.AutoTopicCreationOverrideImpl;
import org.apache.pulsar.common.policies.data.impl.DispatchRateImpl;
@@ -836,9 +837,11 @@ public abstract class NamespacesBase extends AdminResource
{
"Invalid configuration for autoTopicCreationOverride.
the detail is "
+ validateResult.getErrorInfo());
}
- if (maxPartitions > 0 &&
autoTopicCreationOverride.getDefaultNumPartitions() > maxPartitions) {
- throw new RestException(Status.NOT_ACCEPTABLE,
- "Number of partitions should be less than or equal to
" + maxPartitions);
+ if (Objects.equals(autoTopicCreationOverride.getTopicType(),
TopicType.PARTITIONED.toString())) {
+ if (maxPartitions > 0 &&
autoTopicCreationOverride.getDefaultNumPartitions() > maxPartitions) {
+ throw new RestException(Status.NOT_ACCEPTABLE,
+ "Number of partitions should be less than or equal
to " + maxPartitions);
+ }
}
}
// Force to read the data s.t. the watch to the cache content is setup.
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerWrapper.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerWrapper.java
index d18cd616586..d8966a44d4c 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerWrapper.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerWrapper.java
@@ -18,8 +18,8 @@
*/
package org.apache.pulsar.broker.loadbalance.impl;
+import java.util.Collections;
import java.util.List;
-import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
@@ -70,7 +70,8 @@ public class ModularLoadManagerWrapper implements LoadManager
{
String webServiceUrl = getBrokerWebServiceUrl(s);
String brokerZnodeName = getBrokerZnodeName(s, webServiceUrl);
return new SimpleResourceUnit(webServiceUrl,
- new PulsarResourceDescription(),
Map.of(ResourceUnit.PROPERTY_KEY_BROKER_ZNODE_NAME, brokerZnodeName));
+ new PulsarResourceDescription(),
+
Collections.singletonMap(ResourceUnit.PROPERTY_KEY_BROKER_ZNODE_NAME,
brokerZnodeName));
});
}
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java
index e8eee716faf..3a2385722e9 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java
@@ -45,6 +45,7 @@ import java.util.Optional;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
+import javax.ws.rs.NotAcceptableException;
import javax.ws.rs.core.Response.Status;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
@@ -84,6 +85,7 @@ import org.apache.pulsar.common.naming.TopicDomain;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.AutoFailoverPolicyData;
import org.apache.pulsar.common.policies.data.AutoFailoverPolicyType;
+import org.apache.pulsar.common.policies.data.AutoTopicCreationOverride;
import org.apache.pulsar.common.policies.data.BacklogQuota;
import org.apache.pulsar.common.policies.data.BrokerNamespaceIsolationData;
import org.apache.pulsar.common.policies.data.BrokerNamespaceIsolationDataImpl;
@@ -1705,6 +1707,48 @@ public class AdminApi2Test extends
MockedPulsarServiceBaseTest {
}
}
+ @Test
+ public void testAutoTopicCreationOverrideWithMaxNumPartitionsLimit()
throws Exception{
+ super.internalCleanup();
+ conf.setMaxNumPartitionsPerPartitionedTopic(10);
+ super.internalSetup();
+ admin.clusters().createCluster("test",
ClusterData.builder().serviceUrl(brokerUrl.toString()).build());
+ TenantInfoImpl tenantInfo = new TenantInfoImpl(
+ Sets.newHashSet("role1", "role2"), Sets.newHashSet("test"));
+ admin.tenants().createTenant("testTenant", tenantInfo);
+ // test non-partitioned
+ admin.namespaces().createNamespace("testTenant/ns1",
Sets.newHashSet("test"));
+ AutoTopicCreationOverride overridePolicy = AutoTopicCreationOverride
+ .builder().allowAutoTopicCreation(true)
+ .topicType("non-partitioned")
+ .build();
+ admin.namespaces().setAutoTopicCreation("testTenant/ns1",
overridePolicy);
+ AutoTopicCreationOverride newOverridePolicy =
+ admin.namespaces().getAutoTopicCreation("testTenant/ns1");
+ assertEquals(overridePolicy, newOverridePolicy);
+ // test partitioned
+ AutoTopicCreationOverride partitionedOverridePolicy =
AutoTopicCreationOverride
+ .builder().allowAutoTopicCreation(true)
+ .topicType("partitioned")
+ .defaultNumPartitions(10)
+ .build();
+ admin.namespaces().setAutoTopicCreation("testTenant/ns1",
partitionedOverridePolicy);
+ AutoTopicCreationOverride partitionedNewOverridePolicy =
+ admin.namespaces().getAutoTopicCreation("testTenant/ns1");
+ assertEquals(partitionedOverridePolicy, partitionedNewOverridePolicy);
+ // test partitioned with error
+ AutoTopicCreationOverride partitionedWrongOverridePolicy =
AutoTopicCreationOverride
+ .builder().allowAutoTopicCreation(true)
+ .topicType("partitioned")
+ .defaultNumPartitions(123)
+ .build();
+ try {
+ admin.namespaces().setAutoTopicCreation("testTenant/ns1",
partitionedWrongOverridePolicy);
+ fail();
+ } catch (Exception ex) {
+ assertTrue(ex.getCause() instanceof NotAcceptableException);
+ }
+ }
@Test
public void testMaxTopicsPerNamespace() throws Exception {
super.internalCleanup();
