[Broker] Make PersistentTopicsBase#internalSetBacklogQuota async
@@ -2860,25 +2860,25 @@ public class PersistentTopicsBase extends AdminResource {
protected CompletableFuture<Void> internalSetBacklogQuota(BacklogQuota.BacklogQuotaType backlogQuotaType,
BacklogQuotaImpl backlogQuota, boolean isGlobal) {
- validateTopicPolicyOperation(topicName, PolicyName.BACKLOG, PolicyOperation.WRITE);
- validatePoliciesReadOnlyAccess();
-
BacklogQuota.BacklogQuotaType finalBacklogQuotaType = backlogQuotaType == null
? BacklogQuota.BacklogQuotaType.destination_storage : backlogQuotaType;
- return getTopicPoliciesAsyncWithRetry(topicName, isGlobal)
+ return validateTopicPolicyOperationAsync(topicName, PolicyName.BACKLOG, PolicyOperation.WRITE)
+ .thenAccept(__ -> validatePoliciesReadOnlyAccess())
+ .thenCompose(__ -> getTopicPoliciesAsyncWithRetry(topicName, isGlobal))
.thenCompose(op -> {
TopicPolicies topicPolicies = op.orElseGet(TopicPolicies::new);
- RetentionPolicies retentionPolicies = getRetentionPolicies(topicName, topicPolicies);
+ return getRetentionPoliciesAsync(topicName, topicPolicies)
+ .thenCompose(retentionPolicies -> {
if (!checkBacklogQuota(backlogQuota, retentionPolicies)) {
log.warn(
- "[{}] Failed to update backlog configuration for topic {}: conflicts with retention quota",
+ "[{}] Failed to update backlog configuration for topic {}: conflicts with"
+ + " retention quota",
clientAppId(), topicName);
return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
"Backlog Quota exceeds configured retention quota for topic. "
+ "Please increase retention quota and retry"));
}
-
if (backlogQuota != null) {
topicPolicies.getBackLogQuotaMap().put(finalBacklogQuotaType.name(), backlogQuota);
} else {
@@ -2886,15 +2886,20 @@ public class PersistentTopicsBase extends AdminResource {
}
Map<String, BacklogQuotaImpl> backLogQuotaMap = topicPolicies.getBackLogQuotaMap();
topicPolicies.setIsGlobal(isGlobal);
- return pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, topicPolicies)
+ return pulsar().getTopicPoliciesService()
+ .updateTopicPoliciesAsync(topicName, topicPolicies)
.thenRun(() -> {
try {
- log.info("[{}] Successfully updated backlog quota map: namespace={}, topic={}, map={}",
+ log.info(
+ "[{}] Successfully updated backlog quota map: namespace={}, "
+ + "topic={}, map={}",
clientAppId(),
namespaceName,
topicName.getLocalName(),
jsonMapper().writeValueAsString(backLogQuotaMap));
- } catch (JsonProcessingException ignore) { }
+ } catch (JsonProcessingException ignore) {
+ }
+ });
});
});
}
@@ -2994,18 +2999,14 @@ public class PersistentTopicsBase extends AdminResource {
});
}
- private RetentionPolicies getRetentionPolicies(TopicName topicName, TopicPolicies topicPolicies) {
+ private CompletableFuture<RetentionPolicies> getRetentionPoliciesAsync(TopicName topicName,
+ TopicPolicies topicPolicies) {
RetentionPolicies retentionPolicies = topicPolicies.getRetentionPolicies();
- if (retentionPolicies == null){
- try {
- retentionPolicies = getNamespacePoliciesAsync(topicName.getNamespaceObject())
- .thenApply(policies -> policies.retention_policies)
- .get(1L, TimeUnit.SECONDS);
- } catch (Exception e) {
- throw new RestException(e);
- }
+ if (retentionPolicies != null) {
+ return CompletableFuture.completedFuture(retentionPolicies);
}
- return retentionPolicies;
+ return getNamespacePoliciesAsync(topicName.getNamespaceObject())
+ .thenApply(policies -> policies.retention_policies);
}
protected CompletableFuture<RetentionPolicies> internalGetRetention(boolean applied, boolean isGlobal) {
