check null to avoid NPE for internalSetTopicPolicies
### Motivation & Changes
1. add null check for topic policies to avoid NPE
2. add boundary check for `MaxUnackedMessagesOnSubscription` and `SetMaxUnackedMessagesOnConsumer`
@@ -899,6 +899,11 @@ public class PersistentTopicsBase extends AdminResource {
}
protected CompletableFuture<Void> internalSetMaxUnackedMessagesOnSubscription(Integer maxUnackedNum) {
+ if (maxUnackedNum != null && maxUnackedNum < 0) {
+ throw new RestException(Status.PRECONDITION_FAILED,
+ "maxUnackedNum must be 0 or more");
+ }
+
TopicPolicies topicPolicies = null;
try {
topicPolicies = pulsar().getTopicPoliciesService().getTopicPolicies(topicName);
@@ -914,6 +919,11 @@ public class PersistentTopicsBase extends AdminResource {
}
protected CompletableFuture<Void> internalSetMaxUnackedMessagesOnConsumer(Integer maxUnackedNum) {
+ if (maxUnackedNum != null && maxUnackedNum < 0) {
+ throw new RestException(Status.PRECONDITION_FAILED,
+ "maxUnackedNum must be 0 or more");
+ }
+
TopicPolicies topicPolicies = null;
try {
topicPolicies = pulsar().getTopicPoliciesService().getTopicPolicies(topicName);
@@ -2308,7 +2318,7 @@ public class PersistentTopicsBase extends AdminResource {
protected void internalSetMessageTTL(AsyncResponse asyncResponse, Integer ttlInSecond) {
//Validate message ttl value.
- if (ttlInSecond != null && ttlInSecond.intValue() < 0) {
+ if (ttlInSecond != null && ttlInSecond < 0) {
throw new RestException(Status.PRECONDITION_FAILED, "Invalid value for message TTL");
}
validateAdminAccessForTenant(namespaceName.getTenant());
@@ -2477,16 +2487,18 @@ public class PersistentTopicsBase extends AdminResource {
}
protected CompletableFuture<Void> internalSetMaxProducers(Integer maxProducers) {
+ if (maxProducers != null && maxProducers < 0) {
+ throw new RestException(Status.PRECONDITION_FAILED,
+ "maxProducers must be 0 or more");
+ }
+
validateAdminAccessForTenant(namespaceName.getTenant());
validatePoliciesReadOnlyAccess();
if (topicName.isGlobal()) {
validateGlobalNamespaceOwnership(namespaceName);
}
checkTopicLevelPolicyEnable();
- if (maxProducers < 0) {
- throw new RestException(Status.PRECONDITION_FAILED,
- "maxProducers must be 0 or more");
- }
+
TopicPolicies topicPolicies = getTopicPolicies(topicName).orElseGet(TopicPolicies::new);
topicPolicies.setMaxProducerPerTopic(maxProducers);
return pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, topicPolicies);
@@ -2518,16 +2530,18 @@ public class PersistentTopicsBase extends AdminResource {
}
protected CompletableFuture<Void> internalSetMaxConsumers(Integer maxConsumers) {
+ if (maxConsumers != null && maxConsumers < 0) {
+ throw new RestException(Status.PRECONDITION_FAILED,
+ "maxConsumers must be 0 or more");
+ }
+
validateAdminAccessForTenant(namespaceName.getTenant());
validatePoliciesReadOnlyAccess();
if (topicName.isGlobal()) {
validateGlobalNamespaceOwnership(namespaceName);
}
checkTopicLevelPolicyEnable();
- if (maxConsumers < 0) {
- throw new RestException(Status.PRECONDITION_FAILED,
- "maxConsumers must be 0 or more");
- }
+
TopicPolicies topicPolicies = getTopicPolicies(topicName).orElseGet(TopicPolicies::new);
topicPolicies.setMaxConsumerPerTopic(maxConsumers);
return pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, topicPolicies);
