lhotari commented on code in PR #24313:
URL: https://github.com/apache/pulsar/pull/24313#discussion_r2093428882
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -568,33 +580,37 @@ protected void
internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
return;
}
validateNamespaceOperationAsync(topicName.getNamespaceObject(),
NamespaceOperation.CREATE_TOPIC)
- .thenRun(() -> {
- Policies policies = null;
- try {
- policies = getNamespacePolicies(namespaceName);
- } catch (RestException e) {
- if (e.getResponse().getStatus() !=
Status.NOT_FOUND.getStatusCode()) {
- throw e;
+ .thenCompose((__) ->
getNamespacePoliciesAsync(namespaceName).exceptionally(ex -> {
+ Throwable unwrapped =
FutureUtil.unwrapCompletionException(ex);
+ if (unwrapped instanceof RestException re) {
+ if (re.getResponse().getStatus() ==
Status.NOT_FOUND.getStatusCode()) {
+ return null;
}
}
-
+ throw new CompletionException(unwrapped);
+ }))
+ .thenCompose(policies -> {
int maxTopicsPerNamespace = policies != null &&
policies.max_topics_per_namespace != null
? policies.max_topics_per_namespace :
pulsar().getConfig().getMaxTopicsPerNamespace();
// new create check
if (maxTopicsPerNamespace > 0 &&
!pulsar().getBrokerService().isSystemTopic(topicName)) {
- List<String> partitionedTopics =
getTopicPartitionList(TopicDomain.persistent);
- // exclude created system topic
- long topicsCount =
- partitionedTopics.stream().filter(t ->
-
!pulsar().getBrokerService().isSystemTopic(TopicName.get(t))).count();
- if (topicsCount + numPartitions >
maxTopicsPerNamespace) {
- log.error("[{}] Failed to create partitioned topic
{}, "
- + "exceed maximum number of topics in
namespace", clientAppId(), topicName);
- throw new RestException(Status.PRECONDITION_FAILED,
- "Exceed maximum number of topics in
namespace.");
- }
+ return
getTopicPartitionListAsync().thenCompose(partitionedTopics -> {
Review Comment:
instead of `.thenCompose`, `.thenRun` could be used here since there's no
asynchronous downstream calls beyond this point. A sign of this is `return
CompletableFuture.completedFuture(null)`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
