Skip to content

Commit cbb1c2a

Browse files
HQebuptHQebupt
committed
[improve][broker] part 1:make some methods async in Namespaces
1 parent f582620 commit cbb1c2a

File tree

4 files changed

+135
-93
lines changed

4 files changed

+135
-93
lines changed

‎pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java

-56
Original file line numberDiff line numberDiff line change
@@ -1293,13 +1293,6 @@ protected CompletableFuture<Void> internalRemovePublishRateAsync() {
12931293
}));
12941294
}
12951295

1296-
protected PublishRate internalGetPublishRate() {
1297-
validateNamespacePolicyOperation(namespaceName, PolicyName.RATE, PolicyOperation.READ);
1298-
1299-
Policies policies = getNamespacePolicies(namespaceName);
1300-
return policies.publishMaxMessageRate.get(pulsar().getConfiguration().getClusterName());
1301-
}
1302-
13031296
protected CompletableFuture<PublishRate> internalGetPublishRateAsync() {
13041297
return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.RATE, PolicyOperation.READ)
13051298
.thenCompose(__ -> getNamespacePoliciesAsync(namespaceName))
@@ -1365,14 +1358,6 @@ protected CompletableFuture<Void> internalDeleteTopicDispatchRateAsync() {
13651358
}));
13661359
}
13671360

1368-
@SuppressWarnings("deprecation")
1369-
protected DispatchRate internalGetTopicDispatchRate() {
1370-
validateNamespacePolicyOperation(namespaceName, PolicyName.RATE, PolicyOperation.READ);
1371-
1372-
Policies policies = getNamespacePolicies(namespaceName);
1373-
return policies.topicDispatchRate.get(pulsar().getConfiguration().getClusterName());
1374-
}
1375-
13761361
@SuppressWarnings("deprecation")
13771362
protected CompletableFuture<DispatchRate> internalGetTopicDispatchRateAsync() {
13781363
return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.RATE, PolicyOperation.READ)
@@ -1825,11 +1810,6 @@ protected Boolean internalGetEncryptionRequired() {
18251810
return policies.encryption_required;
18261811
}
18271812

1828-
protected DelayedDeliveryPolicies internalGetDelayedDelivery() {
1829-
validateNamespacePolicyOperation(namespaceName, PolicyName.DELAYED_DELIVERY, PolicyOperation.READ);
1830-
return getNamespacePolicies(namespaceName).delayed_delivery_policies;
1831-
}
1832-
18331813
protected void internalSetInactiveTopic(InactiveTopicPolicies inactiveTopicPolicies) {
18341814
validateSuperUserAccess();
18351815
validatePoliciesReadOnlyAccess();
@@ -1957,13 +1937,6 @@ protected List<String> internalGetAntiAffinityNamespaces(String cluster, String
19571937
}
19581938
}
19591939

1960-
protected RetentionPolicies internalGetRetention() {
1961-
validateNamespacePolicyOperation(namespaceName, PolicyName.RETENTION, PolicyOperation.READ);
1962-
1963-
Policies policies = getNamespacePolicies(namespaceName);
1964-
return policies.retention_policies;
1965-
}
1966-
19671940
private boolean checkQuotas(Policies policies, RetentionPolicies retention) {
19681941
Map<BacklogQuota.BacklogQuotaType, BacklogQuota> backlogQuotaMap = policies.backlog_quota_map;
19691942
if (backlogQuotaMap.isEmpty()) {
@@ -2244,11 +2217,6 @@ protected void internalSetMaxConsumersPerSubscription(Integer maxConsumersPerSub
22442217
}
22452218
}
22462219

2247-
protected Integer internalGetMaxUnackedMessagesPerConsumer() {
2248-
validateNamespacePolicyOperation(namespaceName, PolicyName.MAX_UNACKED, PolicyOperation.READ);
2249-
return getNamespacePolicies(namespaceName).max_unacked_messages_per_consumer;
2250-
}
2251-
22522220
protected void internalSetMaxUnackedMessagesPerConsumer(Integer maxUnackedMessagesPerConsumer) {
22532221
validateNamespacePolicyOperation(namespaceName, PolicyName.MAX_UNACKED, PolicyOperation.WRITE);
22542222
validatePoliciesReadOnlyAccess();
@@ -2273,16 +2241,6 @@ protected void internalSetMaxUnackedMessagesPerConsumer(Integer maxUnackedMessag
22732241
}
22742242
}
22752243

2276-
protected Integer internalGetMaxUnackedMessagesPerSubscription() {
2277-
validateNamespacePolicyOperation(namespaceName, PolicyName.MAX_UNACKED, PolicyOperation.READ);
2278-
return getNamespacePolicies(namespaceName).max_unacked_messages_per_subscription;
2279-
}
2280-
2281-
protected Integer internalGetMaxSubscriptionsPerTopic() {
2282-
validateNamespacePolicyOperation(namespaceName, PolicyName.MAX_SUBSCRIPTIONS, PolicyOperation.READ);
2283-
return getNamespacePolicies(namespaceName).max_subscriptions_per_topic;
2284-
}
2285-
22862244
protected void internalSetMaxSubscriptionsPerTopic(Integer maxSubscriptionsPerTopic){
22872245
validateSuperUserAccess();
22882246
validatePoliciesReadOnlyAccess();
@@ -2399,14 +2357,6 @@ protected SchemaAutoUpdateCompatibilityStrategy internalGetSchemaAutoUpdateCompa
23992357
return getNamespacePolicies(namespaceName).schema_auto_update_compatibility_strategy;
24002358
}
24012359

2402-
protected SchemaCompatibilityStrategy internalGetSchemaCompatibilityStrategy() {
2403-
validateNamespacePolicyOperation(namespaceName, PolicyName.SCHEMA_COMPATIBILITY_STRATEGY,
2404-
PolicyOperation.READ);
2405-
Policies policies = getNamespacePolicies(namespaceName);
2406-
2407-
return policies.schema_compatibility_strategy;
2408-
}
2409-
24102360
@Deprecated
24112361
protected void internalSetSchemaAutoUpdateCompatibilityStrategy(SchemaAutoUpdateCompatibilityStrategy strategy) {
24122362
validateNamespacePolicyOperation(namespaceName, PolicyName.SCHEMA_COMPATIBILITY_STRATEGY,
@@ -2615,12 +2565,6 @@ private void validateOffloadPolicies(OffloadPoliciesImpl offloadPolicies) {
26152565
}
26162566
}
26172567

2618-
protected int internalGetMaxTopicsPerNamespace() {
2619-
validateNamespacePolicyOperation(namespaceName, PolicyName.MAX_TOPICS, PolicyOperation.READ);
2620-
return getNamespacePolicies(namespaceName).max_topics_per_namespace != null
2621-
? getNamespacePolicies(namespaceName).max_topics_per_namespace : 0;
2622-
}
2623-
26242568
protected void internalRemoveMaxTopicsPerNamespace() {
26252569
validateNamespacePolicyOperation(namespaceName, PolicyName.MAX_TOPICS, PolicyOperation.WRITE);
26262570
internalSetMaxTopicsPerNamespace(null);

‎pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/Namespaces.java

+27-7
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,6 @@
5656
import org.apache.pulsar.common.policies.data.BacklogQuota.BacklogQuotaType;
5757
import org.apache.pulsar.common.policies.data.BookieAffinityGroupData;
5858
import org.apache.pulsar.common.policies.data.BundlesData;
59-
import org.apache.pulsar.common.policies.data.DispatchRate;
6059
import org.apache.pulsar.common.policies.data.NamespaceOperation;
6160
import org.apache.pulsar.common.policies.data.PersistencePolicies;
6261
import org.apache.pulsar.common.policies.data.Policies;
@@ -1017,10 +1016,21 @@ public void setDispatchRate(@Suspended AsyncResponse asyncResponse,
10171016
+ "-1 means msg-dispatch-rate or byte-dispatch-rate not configured in dispatch-rate yet")
10181017
@ApiResponses(value = {@ApiResponse(code = 403, message = "Don't have admin permission"),
10191018
@ApiResponse(code = 404, message = "Namespace does not exist")})
1020-
public DispatchRate getDispatchRate(@PathParam("property") String property, @PathParam("cluster") String cluster,
1021-
@PathParam("namespace") String namespace) {
1019+
public void getDispatchRate(@Suspended final AsyncResponse asyncResponse,
1020+
@PathParam("property") String property,
1021+
@PathParam("cluster") String cluster,
1022+
@PathParam("namespace") String namespace) {
10221023
validateNamespaceName(property, cluster, namespace);
1023-
return internalGetTopicDispatchRate();
1024+
validateNamespacePolicyOperationAsync(namespaceName, PolicyName.RATE, PolicyOperation.READ)
1025+
.thenCompose(__ -> getNamespacePoliciesAsync(namespaceName))
1026+
.thenAccept(policies -> asyncResponse.resume(
1027+
policies.topicDispatchRate.get(pulsar().getConfiguration().getClusterName())))
1028+
.exceptionally(ex -> {
1029+
log.error("[{}] Failed to get dispatch-rate configured for the namespace {}", clientAppId(),
1030+
namespaceName, ex);
1031+
resumeAsyncResponseExceptionally(asyncResponse, ex);
1032+
return null;
1033+
});
10241034
}
10251035

10261036
@POST
@@ -1177,10 +1187,20 @@ public void removeBacklogQuota(@PathParam("property") String property, @PathPara
11771187
@ApiOperation(hidden = true, value = "Get retention config on a namespace.")
11781188
@ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
11791189
@ApiResponse(code = 404, message = "Namespace does not exist") })
1180-
public RetentionPolicies getRetention(@PathParam("property") String property, @PathParam("cluster") String cluster,
1181-
@PathParam("namespace") String namespace) {
1190+
public void getRetention(@Suspended final AsyncResponse asyncResponse,
1191+
@PathParam("property") String property,
1192+
@PathParam("cluster") String cluster,
1193+
@PathParam("namespace") String namespace) {
11821194
validateNamespaceName(property, cluster, namespace);
1183-
return internalGetRetention();
1195+
validateNamespacePolicyOperationAsync(namespaceName, PolicyName.RETENTION, PolicyOperation.READ)
1196+
.thenCompose(__ -> getNamespacePoliciesAsync(namespaceName))
1197+
.thenAccept(policies -> asyncResponse.resume(policies.retention_policies))
1198+
.exceptionally(ex -> {
1199+
log.error("[{}] Failed to get retention config on a namespace {}", clientAppId(), namespaceName,
1200+
ex);
1201+
resumeAsyncResponseExceptionally(asyncResponse, ex);
1202+
return null;
1203+
});
11841204
}
11851205

11861206
@POST

0 commit comments

Comments
 (0)