You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by pe...@apache.org on 2021/06/29 07:24:37 UTC
[pulsar] branch branch-2.7 updated: [Branch-2.7] Add authoritative
flag for topic policy to avoid redirect loop (#11131)
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.7 by this push:
new 33fc3c0 [Branch-2.7] Add authoritative flag for topic policy to avoid redirect loop (#11131)
33fc3c0 is described below
commit 33fc3c0f5d0acad4aabe3bd9f32dcdb83f08e8b0
Author: lipenghui <pe...@apache.org>
AuthorDate: Tue Jun 29 15:23:48 2021 +0800
[Branch-2.7] Add authoritative flag for topic policy to avoid redirect loop (#11131)
* [Branch-2.7] Add authoritative flag for topic policy to avoid redirect loop
Motivation
1. Add authoritative flag for topic policy to avoid redirect loop
2. Prevent set topic policy on a non-existing topic
3. Prevent set topic policy on a partition of a partitioned topic
4. Redirect to the broker which is the owner of the partition-0 for a partitioned topic when setting topic policy
---
.../broker/admin/impl/PersistentTopicsBase.java | 170 +-
.../pulsar/broker/admin/v2/PersistentTopics.java | 1849 +++++++++++---------
.../pulsar/broker/namespace/NamespaceService.java | 27 +-
.../prometheus/PrometheusMetricsGenerator.java | 7 +-
.../pulsar/broker/web/PulsarWebResource.java | 55 +
.../common/naming/NamespaceBundleFactory.java | 2 +-
.../apache/pulsar/broker/admin/AdminApiTest2.java | 2 -
.../pulsar/broker/admin/TopicPoliciesTest.java | 33 +-
.../pulsar/broker/service/BrokerServiceTest.java | 7 +-
.../client/api/SimpleProducerConsumerTest.java | 2 +-
.../discovery/service/DiscoveryServiceTest.java | 3 +-
11 files changed, 1203 insertions(+), 954 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
index 541a2b9..a1c0d63 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
@@ -587,29 +587,17 @@ public class PersistentTopicsBase extends AdminResource {
});
}
- protected void internalSetDelayedDeliveryPolicies(AsyncResponse asyncResponse, DelayedDeliveryPolicies deliveryPolicies) {
- TopicPolicies topicPolicies = null;
+ protected CompletableFuture<Void> internalSetDelayedDeliveryPolicies(DelayedDeliveryPolicies deliveryPolicies) {
+ TopicPolicies topicPolicies;
try {
- topicPolicies = pulsar().getTopicPoliciesService().getTopicPolicies(topicName);
- } catch (BrokerServiceException.TopicPoliciesCacheNotInitException e) {
- log.error("Topic {} policies cache have not init.", topicName);
- asyncResponse.resume(new RestException(e));
- return;
- }
- if (topicPolicies == null) {
- topicPolicies = new TopicPolicies();
+ topicPolicies = getTopicPolicies(topicName).orElseGet(TopicPolicies::new);
+ topicPolicies.setDelayedDeliveryEnabled(deliveryPolicies == null ? null : deliveryPolicies.isActive());
+ topicPolicies.setDelayedDeliveryTickTimeMillis(
+ deliveryPolicies == null ? null : deliveryPolicies.getTickTime());
+ } catch (Exception e) {
+ return FutureUtil.failedFuture(e);
}
- topicPolicies.setDelayedDeliveryEnabled(deliveryPolicies == null ? null : deliveryPolicies.isActive());
- topicPolicies.setDelayedDeliveryTickTimeMillis(deliveryPolicies == null ? null : deliveryPolicies.getTickTime());
- pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, topicPolicies)
- .whenComplete((result, ex) -> {
- if (ex != null) {
- log.error("Failed set delayed delivery policy for topic", ex);
- asyncResponse.resume(new RestException(ex));
- } else {
- asyncResponse.resume(Response.noContent().build());
- }
- });
+ return pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, topicPolicies);
}
private CompletableFuture<Void> updatePartitionInOtherCluster(int numPartitions, Set<String> clusters) {
@@ -2405,54 +2393,44 @@ public class PersistentTopicsBase extends AdminResource {
return offlineTopicStats;
}
- protected void internalSetBacklogQuota(AsyncResponse asyncResponse, BacklogQuota.BacklogQuotaType backlogQuotaType, BacklogQuota backlogQuota) {
+ protected CompletableFuture<Void> internalSetBacklogQuota(BacklogQuota.BacklogQuotaType backlogQuotaType,
+ BacklogQuota backlogQuota) {
+ validatePoliciesReadOnlyAccess();
+ TopicPolicies topicPolicies;
if (backlogQuotaType == null) {
backlogQuotaType = BacklogQuota.BacklogQuotaType.destination_storage;
}
- TopicPolicies topicPolicies;
try {
- topicPolicies = pulsar().getTopicPoliciesService().getTopicPolicies(topicName);
- } catch (BrokerServiceException.TopicPoliciesCacheNotInitException e) {
- log.warn("Topic {} policies cache have not init.", topicName);
- asyncResponse.resume(new RestException(e));
- return;
- }
- if (topicPolicies == null){
- topicPolicies = new TopicPolicies();
+ topicPolicies = getTopicPolicies(topicName).orElseGet(TopicPolicies::new);
+ } catch (Exception e) {
+ return FutureUtil.failedFuture(e);
}
RetentionPolicies retentionPolicies = getRetentionPolicies(topicName, topicPolicies);
- if(!checkBacklogQuota(backlogQuota,retentionPolicies)){
+ if (!checkBacklogQuota(backlogQuota, retentionPolicies)) {
log.warn(
"[{}] Failed to update backlog configuration for topic {}: conflicts with retention quota",
clientAppId(), topicName);
- asyncResponse.resume(new RestException(Status.PRECONDITION_FAILED,
- "Backlog Quota exceeds configured retention quota for topic. " +
- "Please increase retention quota and retry"));
+ 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){
+ if (backlogQuota != null) {
topicPolicies.getBackLogQuotaMap().put(backlogQuotaType.name(), backlogQuota);
- }else {
+ } else {
topicPolicies.getBackLogQuotaMap().remove(backlogQuotaType.name());
}
Map<String, BacklogQuota> backLogQuotaMap = topicPolicies.getBackLogQuotaMap();
- pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, topicPolicies)
- .whenComplete((r, ex) -> {
- if (ex != null) {
- log.error("Failed updated backlog quota map",ex);
- asyncResponse.resume(new RestException(ex));
- } else {
- try {
- log.info("[{}] Successfully updated backlog quota map: namespace={}, topic={}, map={}",
- clientAppId(),
- namespaceName,
- topicName.getLocalName(),
- jsonMapper().writeValueAsString(backLogQuotaMap));
- } catch (JsonProcessingException ignore) { }
- asyncResponse.resume(Response.noContent().build());
- }
- });
+ return pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, topicPolicies).thenRun(() -> {
+ try {
+ log.info("[{}] Successfully updated backlog quota map: namespace={}, topic={}, map={}",
+ clientAppId(),
+ namespaceName,
+ topicName.getLocalName(),
+ jsonMapper().writeValueAsString(backLogQuotaMap));
+ } catch (JsonProcessingException ignore) { }
+ });
}
protected CompletableFuture<Void> internalSetDeduplicationEnabled(Boolean enabled) {
@@ -2470,38 +2448,26 @@ public class PersistentTopicsBase extends AdminResource {
return pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, topicPolicies);
}
- protected void internalSetMessageTTL(AsyncResponse asyncResponse, Integer ttlInSecond) {
+ protected CompletableFuture<Void> internalSetMessageTTL(Integer ttlInSecond) {
//Validate message ttl value.
if (ttlInSecond != null && ttlInSecond < 0) {
- throw new RestException(Status.PRECONDITION_FAILED, "Invalid value for message TTL");
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "Invalid value for message TTL"));
}
TopicPolicies topicPolicies;
- //Update existing topic policy or create a new one if not exist.
try {
- topicPolicies = pulsar().getTopicPoliciesService().getTopicPolicies(topicName);
- } catch (BrokerServiceException.TopicPoliciesCacheNotInitException e) {
- log.warn("Topic {} policies cache have not init.", topicName);
- asyncResponse.resume(new RestException(e));
- return;
- }
- if (topicPolicies == null){
- topicPolicies = new TopicPolicies();
+ topicPolicies = getTopicPolicies(topicName).orElseGet(TopicPolicies::new);
+ } catch (Exception e) {
+ return FutureUtil.failedFuture(e);
}
topicPolicies.setMessageTTLInSeconds(ttlInSecond);
- pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, topicPolicies)
- .whenComplete((result, ex) -> {
- if (ex != null) {
- log.error("Failed set message ttl for topic",ex);
- asyncResponse.resume(new RestException(ex));
- } else {
- log.info("[{}] Successfully set topic message ttl: namespace={}, topic={}, ttl={}",
- clientAppId(),
- namespaceName,
- topicName.getLocalName(),
- ttlInSecond);
- asyncResponse.resume(Response.noContent().build());
- }
- });
+ return pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, topicPolicies).thenRun(() -> {
+ log.info("[{}] Successfully set topic message ttl: namespace={}, topic={}, ttl={}",
+ clientAppId(),
+ namespaceName,
+ topicName.getLocalName(),
+ ttlInSecond);
+ });
}
@@ -2519,19 +2485,9 @@ public class PersistentTopicsBase extends AdminResource {
return retentionPolicies;
}
- protected void internalRemoveBacklogQuota(AsyncResponse asyncResponse,
- BacklogQuota.BacklogQuotaType backlogQuotaType) {
- internalSetBacklogQuota(asyncResponse, backlogQuotaType, null);
- }
-
- protected void internalGetRetention(AsyncResponse asyncResponse){
- Optional<RetentionPolicies> retention = getTopicPolicies(topicName)
- .map(TopicPolicies::getRetentionPolicies);
- if (!retention.isPresent()) {
- asyncResponse.resume(Response.noContent().build());
- }else {
- asyncResponse.resume(retention.get());
- }
+ protected RetentionPolicies internalGetRetention(){
+ return getTopicPolicies(topicName)
+ .map(TopicPolicies::getRetentionPolicies).orElse(new RetentionPolicies());
}
protected CompletableFuture<Void> internalSetRetention(RetentionPolicies retention) {
@@ -2633,12 +2589,30 @@ public class PersistentTopicsBase extends AdminResource {
return pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, topicPolicies);
}
- protected void preValidation() {
+ protected CompletableFuture<Void> preValidation(boolean authoritative) {
checkTopicLevelPolicyEnable();
+ if (topicName.isPartitioned()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "Not allowed to set/get topic policy for a partition"));
+ }
if (topicName.isGlobal()) {
validateGlobalNamespaceOwnership(namespaceName);
}
- validateTopicOwnership(topicName, false);
+ return checkTopicExistsAsync(topicName).thenCompose(exist -> {
+ if (!exist) {
+ throw new RestException(Status.NOT_FOUND, "Topic not found");
+ } else {
+ return getPartitionedTopicMetadataAsync(topicName, false, false)
+ .thenCompose(metadata -> {
+ if (metadata.partitions > 0) {
+ return validateTopicOwnershipAsync(TopicName.get(topicName.toString()
+ + TopicName.PARTITIONED_TOPIC_SUFFIX + 0), authoritative);
+ } else {
+ return validateTopicOwnershipAsync(topicName, authoritative);
+ }
+ });
+ }
+ });
}
protected CompletableFuture<Void> internalRemoveMaxProducers() {
@@ -3612,4 +3586,14 @@ public class PersistentTopicsBase extends AdminResource {
return pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, topicPolicies.get());
}
+ protected void handleTopicPolicyException(String methodName, Throwable thr, AsyncResponse asyncResponse) {
+ Throwable cause = thr.getCause();
+ if (!(cause instanceof WebApplicationException)
+ || !(((WebApplicationException) cause).getResponse().getStatus() == 307)) {
+ log.error("[{}] Failed to perform {} on topic {}",
+ clientAppId(), methodName, topicName, cause);
+ }
+ resumeAsyncResponseExceptionally(asyncResponse, cause);
+ }
+
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
index 45b1b30..d7a1805 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
@@ -261,17 +261,26 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 404, message = "Tenant or cluster or namespace or topic doesn't exist"),
@ApiResponse(code = 500, message = "Internal server error"),})
public void getOffloadPolicies(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- TopicPolicies topicPolicies = getTopicPolicies(topicName).orElse(new TopicPolicies());
- if (topicPolicies.isOffloadPoliciesSet()) {
- asyncResponse.resume(topicPolicies.getOffloadPolicies());
- } else {
- asyncResponse.resume(Response.noContent().build());
- }
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenRun(() -> {
+ TopicPolicies topicPolicies = getTopicPolicies(topicName).orElse(new TopicPolicies());
+ if (topicPolicies.isOffloadPoliciesSet()) {
+ asyncResponse.resume(topicPolicies.getOffloadPolicies());
+ } else {
+ asyncResponse.resume(Response.noContent().build());
+ }
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("getOffloadPolicies", ex, asyncResponse);
+ return null;
+ });
+
}
@POST
@@ -280,24 +289,21 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponses(value = {@ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Tenant or cluster or namespace or topic doesn't exist"),})
public void setOffloadPolicies(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic,
- @ApiParam(value = "Offload policies for the specified topic")
- OffloadPolicies offloadPolicies) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalSetOffloadPolicies(offloadPolicies).whenComplete((res, ex) -> {
- if (ex instanceof RestException) {
- log.error("Failed set offloadPolicies", ex);
- asyncResponse.resume(ex);
- } else if (ex != null) {
- log.error("Failed set offloadPolicies", ex);
- asyncResponse.resume(new RestException(ex));
- } else {
- asyncResponse.resume(Response.noContent().build());
- }
- });
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
+ @ApiParam(value = "Offload policies for the specified topic")
+ OffloadPolicies offloadPolicies) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalSetOffloadPolicies(offloadPolicies))
+ .thenRun(() -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ handleTopicPolicyException("setOffloadPolicies", ex, asyncResponse);
+ return null;
+ });
}
@DELETE
@@ -306,12 +312,19 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponses(value = {@ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Tenant or cluster or namespace or topic doesn't exist"),})
public void removeOffloadPolicies(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- setOffloadPolicies(asyncResponse, tenant, namespace, encodedTopic, null);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalSetOffloadPolicies(null))
+ .thenRun(() -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ handleTopicPolicyException("removeOffloadPolicies", ex, asyncResponse);
+ return null;
+ });
}
@GET
@@ -321,17 +334,25 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 404, message = "Tenant or cluster or namespace or topic doesn't exist"),
@ApiResponse(code = 500, message = "Internal server error"),})
public void getMaxUnackedMessagesOnConsumer(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- TopicPolicies topicPolicies = getTopicPolicies(topicName).orElse(new TopicPolicies());
- if (topicPolicies.isMaxUnackedMessagesOnConsumerSet()) {
- asyncResponse.resume(topicPolicies.getMaxUnackedMessagesOnConsumer());
- } else {
- asyncResponse.resume(Response.noContent().build());
- }
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenRun(() -> {
+ TopicPolicies topicPolicies = getTopicPolicies(topicName).orElse(new TopicPolicies());
+ if (topicPolicies.isMaxUnackedMessagesOnConsumerSet()) {
+ asyncResponse.resume(topicPolicies.getMaxUnackedMessagesOnConsumer());
+ } else {
+ asyncResponse.resume(Response.noContent().build());
+ }
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("getMaxUnackedMessagesOnConsumer", ex, asyncResponse);
+ return null;
+ });
}
@POST
@@ -340,24 +361,42 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponses(value = {@ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Tenant or cluster or namespace or topic doesn't exist"),})
public void setMaxUnackedMessagesOnConsumer(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic,
- @ApiParam(value = "Max unacked messages on consumer policies for the specified topic")
- Integer maxUnackedNum) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalSetMaxUnackedMessagesOnConsumer(maxUnackedNum).whenComplete((res, ex) -> {
- if (ex instanceof RestException) {
- log.error("Failed set MaxUnackedMessagesOnConsumer", ex);
- asyncResponse.resume(ex);
- } else if (ex != null) {
- log.error("Failed set MaxUnackedMessagesOnConsumer", ex);
- asyncResponse.resume(new RestException(ex));
- } else {
- asyncResponse.resume(Response.noContent().build());
- }
- });
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
+ @ApiParam(value = "Max unacked messages on consumer policies for the specified topic")
+ Integer maxUnackedNum) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalSetMaxUnackedMessagesOnConsumer(maxUnackedNum))
+ .thenRun(() -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ handleTopicPolicyException("setMaxUnackedMessagesOnConsumer", ex, asyncResponse);
+ return null;
+ });
+ }
+
+ @DELETE
+ @Path("/{tenant}/{namespace}/{topic}/maxUnackedMessagesOnConsumer")
+ @ApiOperation(value = "Delete max unacked messages per consumer config on a topic.")
+ @ApiResponses(value = {@ApiResponse(code = 403, message = "Don't have admin permission"),
+ @ApiResponse(code = 404, message = "Tenant or cluster or namespace or topic doesn't exist"),})
+ public void deleteMaxUnackedMessagesOnConsumer(@Suspended final AsyncResponse asyncResponse,
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalSetMaxUnackedMessagesOnConsumer(null))
+ .thenRun(() -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ handleTopicPolicyException("deleteMaxUnackedMessagesOnConsumer", ex, asyncResponse);
+ return null;
+ });
}
@GET
@@ -367,17 +406,25 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 404, message = "Tenant or cluster or namespace or topic doesn't exist"),
@ApiResponse(code = 500, message = "Internal server error"),})
public void getDeduplicationSnapshotInterval(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- TopicPolicies topicPolicies = getTopicPolicies(topicName).orElse(new TopicPolicies());
- if (topicPolicies.isDeduplicationSnapshotIntervalSecondsSet()) {
- asyncResponse.resume(topicPolicies.getDeduplicationSnapshotIntervalSeconds());
- } else {
- asyncResponse.resume(Response.noContent().build());
- }
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenRun(() -> {
+ TopicPolicies topicPolicies = getTopicPolicies(topicName).orElse(new TopicPolicies());
+ if (topicPolicies.isDeduplicationSnapshotIntervalSecondsSet()) {
+ asyncResponse.resume(topicPolicies.getDeduplicationSnapshotIntervalSeconds());
+ } else {
+ asyncResponse.resume(Response.noContent().build());
+ }
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("getDeduplicationSnapshotInterval", ex, asyncResponse);
+ return null;
+ });
}
@POST
@@ -386,27 +433,21 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponses(value = {@ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Tenant or cluster or namespace or topic doesn't exist"),})
public void setDeduplicationSnapshotInterval(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic,
- @ApiParam(value = "Interval to take deduplication snapshot for the specified topic")
- Integer interval) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- if (topicName.isGlobal()) {
- validateGlobalNamespaceOwnership(namespaceName);
- }
- internalSetDeduplicationSnapshotInterval(interval).whenComplete((res, ex) -> {
- if (ex instanceof RestException) {
- log.error("Failed set deduplicationSnapshotInterval", ex);
- asyncResponse.resume(ex);
- } else if (ex != null) {
- log.error("Failed set deduplicationSnapshotInterval", ex);
- asyncResponse.resume(new RestException(ex));
- } else {
- asyncResponse.resume(Response.noContent().build());
- }
- });
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
+ @ApiParam(value = "Interval to take deduplication snapshot for the specified topic")
+ Integer interval) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalSetDeduplicationSnapshotInterval(interval))
+ .thenRun(() -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ handleTopicPolicyException("setDeduplicationSnapshotInterval", ex, asyncResponse);
+ return null;
+ });
}
@DELETE
@@ -415,35 +456,19 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponses(value = {@ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Tenant or cluster or namespace or topic doesn't exist"),})
public void deleteDeduplicationSnapshotInterval(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalSetDeduplicationSnapshotInterval(null).whenComplete((res, ex) -> {
- if (ex instanceof RestException) {
- log.error("Failed delete deduplicationSnapshotInterval", ex);
- asyncResponse.resume(ex);
- } else if (ex != null) {
- log.error("Failed delete deduplicationSnapshotInterval", ex);
- asyncResponse.resume(new RestException(ex));
- } else {
- asyncResponse.resume(Response.noContent().build());
- }
- });
- }
-
- @DELETE
- @Path("/{tenant}/{namespace}/{topic}/maxUnackedMessagesOnConsumer")
- @ApiOperation(value = "Delete max unacked messages per consumer config on a topic.")
- @ApiResponses(value = {@ApiResponse(code = 403, message = "Don't have admin permission"),
- @ApiResponse(code = 404, message = "Tenant or cluster or namespace or topic doesn't exist"),})
- public void deleteMaxUnackedMessagesOnConsumer(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateTopicName(tenant, namespace, encodedTopic);
- setMaxUnackedMessagesOnConsumer(asyncResponse, tenant, namespace, encodedTopic, null);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalSetDeduplicationSnapshotInterval(null))
+ .thenRun(() -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ handleTopicPolicyException("deleteDeduplicationSnapshotInterval", ex, asyncResponse);
+ return null;
+ });
}
@GET
@@ -453,17 +478,25 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 404, message = "Tenant or cluster or namespace or topic doesn't exist"),
@ApiResponse(code = 500, message = "Internal server error"),})
public void getInactiveTopicPolicies(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- TopicPolicies topicPolicies = getTopicPolicies(topicName).orElse(new TopicPolicies());
- if (topicPolicies.isInactiveTopicPoliciesSet()) {
- asyncResponse.resume(topicPolicies.getInactiveTopicPolicies());
- } else {
- asyncResponse.resume(Response.noContent().build());
- }
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenRun(() -> {
+ TopicPolicies topicPolicies = getTopicPolicies(topicName).orElse(new TopicPolicies());
+ if (topicPolicies.isInactiveTopicPoliciesSet()) {
+ asyncResponse.resume(topicPolicies.getInactiveTopicPolicies());
+ } else {
+ asyncResponse.resume(Response.noContent().build());
+ }
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("getInactiveTopicPolicies", ex, asyncResponse);
+ return null;
+ });
}
@POST
@@ -472,24 +505,21 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponses(value = {@ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Tenant or cluster or namespace or topic doesn't exist"),})
public void setInactiveTopicPolicies(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic,
- @ApiParam(value = "inactive topic policies for the specified topic")
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
+ @ApiParam(value = "inactive topic policies for the specified topic")
InactiveTopicPolicies inactiveTopicPolicies) {
validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalSetInactiveTopicPolicies(inactiveTopicPolicies).whenComplete((res, ex) -> {
- if (ex instanceof RestException) {
- log.error("Failed set InactiveTopicPolicies", ex);
- asyncResponse.resume(ex);
- } else if (ex != null) {
- log.error("Failed set InactiveTopicPolicies", ex);
- asyncResponse.resume(new RestException(ex));
- } else {
- asyncResponse.resume(Response.noContent().build());
- }
- });
+ preValidation(authoritative)
+ .thenCompose(__ -> internalSetInactiveTopicPolicies(inactiveTopicPolicies))
+ .thenRun(() -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ handleTopicPolicyException("setInactiveTopicPolicies", ex, asyncResponse);
+ return null;
+ });
}
@DELETE
@@ -498,11 +528,19 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponses(value = {@ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Tenant or cluster or namespace or topic doesn't exist"),})
public void deleteInactiveTopicPolicies(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateTopicName(tenant, namespace, encodedTopic);
- setInactiveTopicPolicies(asyncResponse, tenant, namespace, encodedTopic, null);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalSetInactiveTopicPolicies(null))
+ .thenRun(() -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ handleTopicPolicyException("deleteInactiveTopicPolicies", ex, asyncResponse);
+ return null;
+ });
}
@GET
@@ -512,17 +550,25 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 404, message = "Tenant or cluster or namespace or topic doesn't exist"),
@ApiResponse(code = 500, message = "Internal server error"),})
public void getMaxUnackedMessagesOnSubscription(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- TopicPolicies topicPolicies = getTopicPolicies(topicName).orElse(new TopicPolicies());
- if (topicPolicies.isMaxUnackedMessagesOnSubscriptionSet()) {
- asyncResponse.resume(topicPolicies.getMaxUnackedMessagesOnSubscription());
- } else {
- asyncResponse.resume(Response.noContent().build());
- }
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenRun(() -> {
+ TopicPolicies topicPolicies = getTopicPolicies(topicName).orElse(new TopicPolicies());
+ if (topicPolicies.isMaxUnackedMessagesOnSubscriptionSet()) {
+ asyncResponse.resume(topicPolicies.getMaxUnackedMessagesOnSubscription());
+ } else {
+ asyncResponse.resume(Response.noContent().build());
+ }
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("getInactiveTopicPolicies", ex, asyncResponse);
+ return null;
+ });
}
@POST
@@ -531,24 +577,21 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponses(value = {@ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Tenant or cluster or namespace or topic doesn't exist"),})
public void setMaxUnackedMessagesOnSubscription(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic,
- @ApiParam(value = "Max unacked messages on subscription policies for the specified topic")
- Integer maxUnackedNum) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalSetMaxUnackedMessagesOnSubscription(maxUnackedNum).whenComplete((res, ex) -> {
- if (ex instanceof RestException) {
- log.error("Failed set MaxUnackedMessagesOnSubscription", ex);
- asyncResponse.resume(ex);
- } else if (ex != null) {
- log.error("Failed set MaxUnackedMessagesOnSubscription", ex);
- asyncResponse.resume(new RestException(ex));
- } else {
- asyncResponse.resume(Response.noContent().build());
- }
- });
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
+ @ApiParam(value = "Max unacked messages on subscription policies for the specified topic")
+ Integer maxUnackedNum) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalSetMaxUnackedMessagesOnSubscription(maxUnackedNum))
+ .thenRun(() -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ handleTopicPolicyException("setMaxUnackedMessagesOnSubscription", ex, asyncResponse);
+ return null;
+ });
}
@@ -559,11 +602,20 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponses(value = {@ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Tenant or cluster or namespace or topic doesn't exist"),})
public void deleteMaxUnackedMessagesOnSubscription(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateTopicName(tenant, namespace, encodedTopic);
- setMaxUnackedMessagesOnSubscription(asyncResponse, tenant, namespace, encodedTopic, null);
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalSetMaxUnackedMessagesOnSubscription(null))
+ .thenRun(() -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ handleTopicPolicyException("deleteMaxUnackedMessagesOnSubscription", ex, asyncResponse);
+ return null;
+ });
}
@GET
@@ -573,18 +625,26 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 404, message = "Tenant or cluster or namespace or topic doesn't exist"),
@ApiResponse(code = 500, message = "Internal server error"),})
public void getDelayedDeliveryPolicies(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- TopicPolicies topicPolicies = getTopicPolicies(topicName).orElse(new TopicPolicies());
- if (topicPolicies.isDelayedDeliveryEnabledSet() && topicPolicies.isDelayedDeliveryTickTimeMillisSet()) {
- asyncResponse.resume(new DelayedDeliveryPolicies(topicPolicies.getDelayedDeliveryTickTimeMillis()
- , topicPolicies.getDelayedDeliveryEnabled()));
- } else {
- asyncResponse.resume(Response.noContent().build());
- }
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenRun(() -> {
+ TopicPolicies topicPolicies = getTopicPolicies(topicName).orElse(new TopicPolicies());
+ if (topicPolicies.isDelayedDeliveryEnabledSet() && topicPolicies.isDelayedDeliveryTickTimeMillisSet()) {
+ asyncResponse.resume(new DelayedDeliveryPolicies(topicPolicies.getDelayedDeliveryTickTimeMillis()
+ , topicPolicies.getDelayedDeliveryEnabled()));
+ } else {
+ asyncResponse.resume(Response.noContent().build());
+ }
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("getDelayedDeliveryPolicies", ex, asyncResponse);
+ return null;
+ });
}
@POST
@@ -593,13 +653,20 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponses(value = {@ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Tenant or cluster or namespace or topic doesn't exist"),})
public void setDelayedDeliveryPolicies(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic,
- @ApiParam(value = "Delayed delivery policies for the specified topic") DelayedDeliveryPolicies deliveryPolicies) {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
+ @ApiParam(value = "Delayed delivery policies for the specified topic") DelayedDeliveryPolicies deliveryPolicies) {
validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalSetDelayedDeliveryPolicies(asyncResponse, deliveryPolicies);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalSetDelayedDeliveryPolicies(deliveryPolicies))
+ .thenRun(() -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ handleTopicPolicyException("setDelayedDeliveryPolicies", ex, asyncResponse);
+ return null;
+ });
}
@@ -610,11 +677,19 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponses(value = {@ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Tenant or cluster or namespace or topic doesn't exist"),})
public void deleteDelayedDeliveryPolicies(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateTopicName(tenant, namespace, encodedTopic);
- setDelayedDeliveryPolicies(asyncResponse, tenant, namespace, encodedTopic, null);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalSetDelayedDeliveryPolicies(null))
+ .thenRun(() -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ handleTopicPolicyException("deleteDelayedDeliveryPolicies", ex, asyncResponse);
+ return null;
+ });
}
/**
@@ -1454,21 +1529,27 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Topic policy does not exist"),
@ApiResponse(code = 405, message = "Topic level policy is disabled, to enable the topic level policy and retry")})
- public Map<BacklogQuota.BacklogQuotaType, BacklogQuota> getBacklogQuotaMap(@PathParam("tenant") String tenant,
+ public void getBacklogQuotaMap(
+ @Suspended final AsyncResponse asyncResponse,
+ @PathParam("tenant") String tenant,
@PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- return getTopicPolicies(topicName)
+ preValidation(authoritative)
+ .thenRun(() -> asyncResponse.resume(getTopicPolicies(topicName)
.map(TopicPolicies::getBackLogQuotaMap)
.map(map -> {
HashMap<BacklogQuota.BacklogQuotaType, BacklogQuota> hashMap = Maps.newHashMap();
- map.forEach((key,value) -> {
- hashMap.put(BacklogQuota.BacklogQuotaType.valueOf(key),value);
- });
+ map.forEach((key,value) -> hashMap.put(BacklogQuota.BacklogQuotaType.valueOf(key),value));
return hashMap;
})
- .orElse(Maps.newHashMap());
+ .orElse(Maps.newHashMap())))
+ .exceptionally(ex -> {
+ handleTopicPolicyException("getBacklogQuotaMap", ex, asyncResponse);
+ return null;
+ });
}
@POST
@@ -1479,12 +1560,21 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 409, message = "Concurrent modification"),
@ApiResponse(code = 405, message = "Topic level policy is disabled, to enable the topic level policy and retry"),
@ApiResponse(code = 412, message = "Specified backlog quota exceeds retention quota. Increase retention quota and retry request") })
- public void setBacklogQuota(@Suspended final AsyncResponse asyncResponse, @PathParam("tenant") String tenant, @PathParam("namespace") String namespace,
+ public void setBacklogQuota(@Suspended final AsyncResponse asyncResponse,
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
@PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
@QueryParam("backlogQuotaType") BacklogQuota.BacklogQuotaType backlogQuotaType, BacklogQuota backlogQuota) {
validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalSetBacklogQuota(asyncResponse, backlogQuotaType, backlogQuota);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalSetBacklogQuota(backlogQuotaType, backlogQuota))
+ .thenRun(() -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ handleTopicPolicyException("setBacklogQuota", ex, asyncResponse);
+ return null;
+ });
}
@DELETE
@@ -1494,12 +1584,21 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 404, message = "Topic does not exist"),
@ApiResponse(code = 405, message = "Topic level policy is disabled, to enable the topic level policy and retry"),
@ApiResponse(code = 409, message = "Concurrent modification") })
- public void removeBacklogQuota(@Suspended final AsyncResponse asyncResponse, @PathParam("tenant") String tenant, @PathParam("namespace") String namespace,
+ public void removeBacklogQuota(@Suspended final AsyncResponse asyncResponse,
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
@PathParam("topic") @Encoded String encodedTopic,
- @QueryParam("backlogQuotaType") BacklogQuota.BacklogQuotaType backlogQuotaType) {
+ @QueryParam("backlogQuotaType") BacklogQuota.BacklogQuotaType backlogQuotaType,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalRemoveBacklogQuota(asyncResponse, backlogQuotaType);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalSetBacklogQuota(backlogQuotaType, null))
+ .thenRun(() -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ handleTopicPolicyException("removeBacklogQuota", ex, asyncResponse);
+ return null;
+ });
}
@GET
@@ -1508,14 +1607,21 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Topic does not exist"),
@ApiResponse(code = 405, message = "Topic level policy is disabled, enable the topic level policy and retry")})
- public int getMessageTTL(@PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
+ public void getMessageTTL(@Suspended final AsyncResponse asyncResponse,
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- return getTopicPolicies(topicName)
+ preValidation(authoritative)
+ .thenRun(() -> asyncResponse.resume(getTopicPolicies(topicName)
.map(TopicPolicies::getMessageTTLInSeconds)
- .orElse(0); //same as default ttl at namespace level
+ .orElse(0)))
+ .exceptionally(ex -> {
+ handleTopicPolicyException("getMessageTTL", ex, asyncResponse);
+ return null;
+ });
}
@POST
@@ -1526,14 +1632,21 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 405, message = "Topic level policy is disabled, enable the topic level policy and retry"),
@ApiResponse(code = 412, message = "Invalid message TTL value") })
public void setMessageTTL(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic,
- @ApiParam(value = "TTL in seconds for the specified namespace", required = true)
- @QueryParam("messageTTL") int messageTTL) {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "TTL in seconds for the specified namespace", required = true)
+ @QueryParam("messageTTL") int messageTTL,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalSetMessageTTL(asyncResponse, messageTTL);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalSetMessageTTL(messageTTL))
+ .thenRun(() -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ handleTopicPolicyException("setMessageTTL", ex, asyncResponse);
+ return null;
+ });
}
@DELETE
@@ -1544,12 +1657,19 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 405, message = "Topic level policy is disabled, enable the topic level policy and retry"),
@ApiResponse(code = 412, message = "Invalid message TTL value") })
public void removeMessageTTL(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalSetMessageTTL(asyncResponse, null);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalSetMessageTTL(null))
+ .thenRun(() -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ handleTopicPolicyException("removeMessageTTL", ex, asyncResponse);
+ return null;
+ });
}
@GET
@@ -1559,17 +1679,25 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 404, message = "Tenant or cluster or namespace or topic doesn't exist"),
@ApiResponse(code = 405, message = "Topic level policy is disabled, to enable the topic level policy and retry")})
public void getDeduplicationEnabled(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- TopicPolicies topicPolicies = getTopicPolicies(topicName).orElse(new TopicPolicies());
- if (topicPolicies.isDeduplicationSet()) {
- asyncResponse.resume(topicPolicies.getDeduplicationEnabled());
- } else {
- asyncResponse.resume(Response.noContent().build());
- }
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenRun(() -> {
+ TopicPolicies topicPolicies = getTopicPolicies(topicName).orElse(new TopicPolicies());
+ if (topicPolicies.isDeduplicationSet()) {
+ asyncResponse.resume(topicPolicies.getDeduplicationEnabled());
+ } else {
+ asyncResponse.resume(Response.noContent().build());
+ }
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("getDeduplicationEnabled", ex, asyncResponse);
+ return null;
+ });
}
@POST
@@ -1579,23 +1707,20 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 404, message = "Tenant or cluster or namespace or topic doesn't exist"),
@ApiResponse(code = 405, message = "Topic level policy is disabled, to enable the topic level policy and retry")})
public void setDeduplicationEnabled(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic,
- @ApiParam(value = "DeduplicationEnabled policies for the specified topic") Boolean enabled) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalSetDeduplicationEnabled(enabled).whenComplete((r, ex) -> {
- if (ex instanceof RestException) {
- log.error("Failed updated deduplication", ex);
- asyncResponse.resume(ex);
- }else if (ex != null) {
- log.error("Failed updated deduplication", ex);
- asyncResponse.resume(new RestException(ex));
- } else {
- asyncResponse.resume(Response.noContent().build());
- }
- });
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
+ @ApiParam(value = "DeduplicationEnabled policies for the specified topic") Boolean enabled) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalSetDeduplicationEnabled(enabled))
+ .thenRun(() -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ handleTopicPolicyException("setDeduplication", ex, asyncResponse);
+ return null;
+ });
}
@DELETE
@@ -1606,11 +1731,19 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 405, message = "Topic level policy is disabled, to enable the topic level policy and retry"),
@ApiResponse(code = 409, message = "Concurrent modification") })
public void removeDeduplicationEnabled(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateTopicName(tenant, namespace, encodedTopic);
- setDeduplicationEnabled(asyncResponse, tenant, namespace, encodedTopic, null);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalSetDeduplicationEnabled(null))
+ .thenRun(() -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ handleTopicPolicyException("removeDeduplication", ex, asyncResponse);
+ return null;
+ });
}
@GET
@@ -1623,16 +1756,23 @@ public class PersistentTopics extends PersistentTopicsBase {
public void getRetention(@Suspended final AsyncResponse asyncResponse,
@PathParam("tenant") String tenant,
@PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- try {
- internalGetRetention(asyncResponse);
- } catch (RestException e) {
- asyncResponse.resume(e);
- } catch (Exception e) {
- asyncResponse.resume(new RestException(e));
- }
+ preValidation(authoritative)
+ .thenRun(() -> {
+ TopicPolicies topicPolicies = getTopicPolicies(topicName).orElse(new TopicPolicies());
+ if (topicPolicies.isRetentionSet()) {
+ asyncResponse.resume(topicPolicies.getRetentionPolicies());
+ } else {
+ asyncResponse.resume(Response.noContent().build());
+ }
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("getRetention", ex, asyncResponse);
+ return null;
+ });
}
@POST
@@ -1647,27 +1787,27 @@ public class PersistentTopics extends PersistentTopicsBase {
@PathParam("tenant") String tenant,
@PathParam("namespace") String namespace,
@PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
@ApiParam(value = "Retention policies for the specified namespace") RetentionPolicies retention) {
validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalSetRetention(retention).whenComplete((r, ex) -> {
- if (ex instanceof RestException) {
- log.error("Failed updated retention", ex);
- asyncResponse.resume(ex);
- }else if (ex != null) {
- log.error("Failed updated retention", ex);
- asyncResponse.resume(new RestException(ex));
- } else {
+ preValidation(authoritative)
+ .thenCompose(__ -> internalSetRetention(retention))
+ .thenRun(() -> {
try {
log.info("[{}] Successfully updated retention: namespace={}, topic={}, retention={}",
clientAppId(),
namespaceName,
topicName.getLocalName(),
jsonMapper().writeValueAsString(retention));
- } catch (JsonProcessingException ignore) { }
+ } catch (JsonProcessingException ignore) {
+ }
asyncResponse.resume(Response.noContent().build());
- }
- });
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("setRetention", ex, asyncResponse);
+ return null;
+ });
}
@DELETE
@@ -1679,23 +1819,25 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 409, message = "Concurrent modification"),
@ApiResponse(code = 412, message = "Retention Quota must exceed backlog quota") })
public void removeRetention(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalRemoveRetention().whenComplete((r, ex) -> {
- if (ex != null) {
- log.error("Failed updated retention", ex);
- asyncResponse.resume(new RestException(ex));
- } else {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalRemoveRetention())
+ .thenRun(() -> {
log.info("[{}] Successfully remove retention: namespace={}, topic={}",
clientAppId(),
namespaceName,
topicName.getLocalName());
asyncResponse.resume(Response.noContent().build());
- }
- });
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("removeRetention", ex, asyncResponse);
+ return null;
+ });
}
@GET
@@ -1706,23 +1848,25 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 405, message = "Topic level policy is disabled, to enable the topic level policy and retry"),
@ApiResponse(code = 409, message = "Concurrent modification")})
public void getPersistence(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- try {
- Optional<PersistencePolicies> persistencePolicies = internalGetPersistence();
- if (!persistencePolicies.isPresent()) {
- asyncResponse.resume(Response.noContent().build());
- } else {
- asyncResponse.resume(persistencePolicies.get());
- }
- } catch (RestException e) {
- asyncResponse.resume(e);
- } catch (Exception e) {
- asyncResponse.resume(new RestException(e));
- }
+ preValidation(authoritative)
+ .thenRun(() -> {
+ Optional<PersistencePolicies> persistencePolicies = internalGetPersistence();
+ if (!persistencePolicies.isPresent()) {
+ asyncResponse.resume(Response.noContent().build());
+ } else {
+ asyncResponse.resume(persistencePolicies.get());
+ }
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("getPersistence", ex, asyncResponse);
+ return null;
+ });
}
@POST
@@ -1734,22 +1878,19 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 409, message = "Concurrent modification"),
@ApiResponse(code = 400, message = "Invalid persistence policies")})
public void setPersistence(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic,
- @ApiParam(value = "Bookkeeper persistence policies for specified topic") PersistencePolicies persistencePolicies) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalSetPersistence(persistencePolicies).whenComplete((r, ex) -> {
- if (ex instanceof RestException) {
- log.error("Failed updated persistence policies", ex);
- asyncResponse.resume(ex);
- } else if (ex != null) {
- log.error("Failed updated persistence policies", ex);
- asyncResponse.resume(new RestException(ex));
- } else {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
+ @ApiParam(value = "Bookkeeper persistence policies for specified topic") PersistencePolicies persistencePolicies) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalSetPersistence(persistencePolicies))
+ .thenRun(() -> {
try {
- log.info("[{}] Successfully updated persistence policies: namespace={}, topic={}, persistencePolicies={}",
+ log.info("[{}] Successfully updated persistence policies: "
+ + "namespace={}, topic={}, persistencePolicies={}",
clientAppId(),
namespaceName,
topicName.getLocalName(),
@@ -1757,8 +1898,11 @@ public class PersistentTopics extends PersistentTopicsBase {
} catch (JsonProcessingException ignore) {
}
asyncResponse.resume(Response.noContent().build());
- }
- });
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("setPersistence", ex, asyncResponse);
+ return null;
+ });
}
@DELETE
@@ -1769,23 +1913,25 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 405, message = "Topic level policy is disabled, to enable the topic level policy and retry"),
@ApiResponse(code = 409, message = "Concurrent modification")})
public void removePersistence(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalRemovePersistence().whenComplete((r, ex) -> {
- if (ex != null) {
- log.error("Failed updated retention", ex);
- asyncResponse.resume(new RestException(ex));
- } else {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalRemovePersistence())
+ .thenRun(() -> {
log.info("[{}] Successfully remove persistence policies: namespace={}, topic={}",
clientAppId(),
namespaceName,
topicName.getLocalName());
asyncResponse.resume(Response.noContent().build());
- }
- });
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("removePersistence", ex, asyncResponse);
+ return null;
+ });
}
@GET
@@ -1797,23 +1943,25 @@ public class PersistentTopics extends PersistentTopicsBase {
message = "Topic level policy is disabled, to enable the topic level policy and retry"),
@ApiResponse(code = 409, message = "Concurrent modification")})
public void getMaxSubscriptionsPerTopic(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- try {
- Optional<Integer> maxSubscriptionsPerTopic = internalGetMaxSubscriptionsPerTopic();
- if (!maxSubscriptionsPerTopic.isPresent()) {
- asyncResponse.resume(Response.noContent().build());
- } else {
- asyncResponse.resume(maxSubscriptionsPerTopic.get());
- }
- } catch (RestException e) {
- asyncResponse.resume(e);
- } catch (Exception e) {
- asyncResponse.resume(new RestException(e));
- }
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenRun(() -> {
+ Optional<Integer> maxSubscriptionsPerTopic = internalGetMaxSubscriptionsPerTopic();
+ if (!maxSubscriptionsPerTopic.isPresent()) {
+ asyncResponse.resume(Response.noContent().build());
+ } else {
+ asyncResponse.resume(maxSubscriptionsPerTopic.get());
+ }
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("getMaxSubscriptionsPerTopic", ex, asyncResponse);
+ return null;
+ });
}
@POST
@@ -1826,26 +1974,25 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 409, message = "Concurrent modification"),
@ApiResponse(code = 412, message = "Invalid value of maxSubscriptionsPerTopic")})
public void setMaxSubscriptionsPerTopic(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic,
- @ApiParam(value = "The max subscriptions of the topic") int maxSubscriptionsPerTopic) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalSetMaxSubscriptionsPerTopic(maxSubscriptionsPerTopic).whenComplete((r, ex) -> {
- if (ex instanceof RestException) {
- log.error("Updating maxSubscriptionsPerTopic failed", ex);
- asyncResponse.resume(ex);
- } else if (ex != null) {
- log.error("Updating maxSubscriptionsPerTopic failed", ex);
- asyncResponse.resume(new RestException(ex));
- } else {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
+ @ApiParam(value = "The max subscriptions of the topic") int maxSubscriptionsPerTopic) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalSetMaxSubscriptionsPerTopic(maxSubscriptionsPerTopic))
+ .thenRun(() -> {
log.info("[{}] Successfully updated maxSubscriptionsPerTopic: namespace={}, topic={}"
+ ", maxSubscriptions={}"
, clientAppId(), namespaceName, topicName.getLocalName(), maxSubscriptionsPerTopic);
asyncResponse.resume(Response.noContent().build());
- }
- });
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("setMaxSubscriptions", ex, asyncResponse);
+ return null;
+ });
}
@DELETE
@@ -1857,23 +2004,23 @@ public class PersistentTopics extends PersistentTopicsBase {
message = "Topic level policy is disabled, to enable the topic level policy and retry"),
@ApiResponse(code = 409, message = "Concurrent modification")})
public void removeMaxSubscriptionsPerTopic(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalSetMaxSubscriptionsPerTopic(null).whenComplete((r, ex) -> {
- if (ex != null) {
- log.error("Failed to remove maxSubscriptionsPerTopic", ex);
- asyncResponse.resume(new RestException(ex));
- } else {
- log.info("[{}] Successfully remove maximum subscription limit: namespace={}, topic={}",
- clientAppId(),
- namespaceName,
- topicName.getLocalName());
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalSetMaxSubscriptionsPerTopic(null))
+ .thenRun(() -> {
+ log.info("[{}] Successfully remove maxSubscriptionsPerTopic: namespace={}, topic={}",
+ clientAppId(), namespaceName, topicName.getLocalName());
asyncResponse.resume(Response.noContent().build());
- }
- });
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("removeMaxSubscriptions", ex, asyncResponse);
+ return null;
+ });
}
@GET
@@ -1884,23 +2031,25 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 405, message = "Topic level policy is disabled, to enable the topic level policy and retry"),
@ApiResponse(code = 409, message = "Concurrent modification")})
public void getMaxProducers(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- try {
- Optional<Integer> maxProducers = internalGetMaxProducers();
- if (!maxProducers.isPresent()) {
- asyncResponse.resume(Response.noContent().build());
- } else {
- asyncResponse.resume(maxProducers.get());
- }
- } catch (RestException e) {
- asyncResponse.resume(e);
- } catch (Exception e) {
- asyncResponse.resume(new RestException(e));
- }
+ preValidation(authoritative)
+ .thenRun(() -> {
+ Optional<Integer> maxProducers = internalGetMaxProducers();
+ if (!maxProducers.isPresent()) {
+ asyncResponse.resume(Response.noContent().build());
+ } else {
+ asyncResponse.resume(maxProducers.get());
+ }
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("getMaxProducers", ex, asyncResponse);
+ return null;
+ });
}
@POST
@@ -1912,28 +2061,27 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 409, message = "Concurrent modification"),
@ApiResponse(code = 412, message = "Invalid value of maxProducers")})
public void setMaxProducers(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic,
- @ApiParam(value = "The max producers of the topic") int maxProducers) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalSetMaxProducers(maxProducers).whenComplete((r, ex) -> {
- if (ex instanceof RestException) {
- log.error("Failed updated persistence policies", ex);
- asyncResponse.resume(ex);
- } else if (ex != null) {
- log.error("Failed updated persistence policies", ex);
- asyncResponse.resume(new RestException(ex));
- } else {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
+ @ApiParam(value = "The max producers of the topic") int maxProducers) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalSetMaxProducers(maxProducers))
+ .thenRun(() -> {
log.info("[{}] Successfully updated max producers: namespace={}, topic={}, maxProducers={}",
clientAppId(),
namespaceName,
topicName.getLocalName(),
maxProducers);
asyncResponse.resume(Response.noContent().build());
- }
- });
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("setMaxProducers", ex, asyncResponse);
+ return null;
+ });
}
@DELETE
@@ -1944,23 +2092,25 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 405, message = "Topic level policy is disabled, to enable the topic level policy and retry"),
@ApiResponse(code = 409, message = "Concurrent modification")})
public void removeMaxProducers(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalRemoveMaxProducers().whenComplete((r, ex) -> {
- if (ex != null) {
- log.error("Failed to remove maxProducers", ex);
- asyncResponse.resume(new RestException(ex));
- } else {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalRemoveMaxProducers())
+ .thenRun(() -> {
log.info("[{}] Successfully remove max producers: namespace={}, topic={}",
clientAppId(),
namespaceName,
topicName.getLocalName());
asyncResponse.resume(Response.noContent().build());
- }
- });
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("removeMaxProducers", ex, asyncResponse);
+ return null;
+ });
}
@GET
@@ -1971,23 +2121,25 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 405, message = "Topic level policy is disabled, to enable the topic level policy and retry"),
@ApiResponse(code = 409, message = "Concurrent modification")})
public void getMaxConsumers(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- try {
- Optional<Integer> maxConsumers = internalGetMaxConsumers();
- if (!maxConsumers.isPresent()) {
- asyncResponse.resume(Response.noContent().build());
- } else {
- asyncResponse.resume(maxConsumers.get());
- }
- } catch (RestException e) {
- asyncResponse.resume(e);
- } catch (Exception e) {
- asyncResponse.resume(new RestException(e));
- }
+ preValidation(authoritative)
+ .thenRun(() -> {
+ Optional<Integer> maxConsumers = internalGetMaxConsumers();
+ if (!maxConsumers.isPresent()) {
+ asyncResponse.resume(Response.noContent().build());
+ } else {
+ asyncResponse.resume(maxConsumers.get());
+ }
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("getMaxConsumers", ex, asyncResponse);
+ return null;
+ });
}
@POST
@@ -1999,28 +2151,27 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 409, message = "Concurrent modification"),
@ApiResponse(code = 412, message = "Invalid value of maxConsumers")})
public void setMaxConsumers(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic,
- @ApiParam(value = "The max consumers of the topic") int maxConsumers) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalSetMaxConsumers(maxConsumers).whenComplete((r, ex) -> {
- if (ex instanceof RestException) {
- log.error("Failed updated persistence policies", ex);
- asyncResponse.resume(ex);
- } else if (ex != null) {
- log.error("Failed updated persistence policies", ex);
- asyncResponse.resume(new RestException(ex));
- } else {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
+ @ApiParam(value = "The max consumers of the topic") int maxConsumers) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalSetMaxConsumers(maxConsumers))
+ .thenRun(() -> {
log.info("[{}] Successfully updated max consumers: namespace={}, topic={}, maxConsumers={}",
clientAppId(),
namespaceName,
topicName.getLocalName(),
maxConsumers);
asyncResponse.resume(Response.noContent().build());
- }
- });
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("setMaxConsumers", ex, asyncResponse);
+ return null;
+ });
}
@DELETE
@@ -2031,23 +2182,25 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 405, message = "Topic level policy is disabled, to enable the topic level policy and retry"),
@ApiResponse(code = 409, message = "Concurrent modification")})
public void removeMaxConsumers(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalRemoveMaxConsumers().whenComplete((r, ex) -> {
- if (ex != null) {
- log.error("Failed to remove maxConsumers", ex);
- asyncResponse.resume(new RestException(ex));
- } else {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalRemoveMaxConsumers())
+ .thenRun(() -> {
log.info("[{}] Successfully remove max consumers: namespace={}, topic={}",
clientAppId(),
namespaceName,
topicName.getLocalName());
asyncResponse.resume(Response.noContent().build());
- }
- });
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("removeMaxConsumers", ex, asyncResponse);
+ return null;
+ });
}
@GET
@@ -2058,23 +2211,25 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 405, message = "Topic level policy is disabled, to enable the topic level policy and retry"),
@ApiResponse(code = 409, message = "Concurrent modification")})
public void getMaxMessageSize(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- try {
- Optional<Integer> policies = internalGetMaxMessageSize();
- if (policies.isPresent()) {
- asyncResponse.resume(policies.get());
- } else {
- asyncResponse.resume(Response.noContent().build());
- }
- } catch (RestException e) {
- asyncResponse.resume(e);
- } catch (Exception e) {
- asyncResponse.resume(new RestException(e));
- }
+ preValidation(authoritative)
+ .thenRun(() -> {
+ Optional<Integer> policies = internalGetMaxMessageSize();
+ if (policies.isPresent()) {
+ asyncResponse.resume(policies.get());
+ } else {
+ asyncResponse.resume(Response.noContent().build());
+ }
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("getMaxMessageSize", ex, asyncResponse);
+ return null;
+ });
}
@POST
@@ -2086,28 +2241,27 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 409, message = "Concurrent modification"),
@ApiResponse(code = 412, message = "Invalid value of maxConsumers")})
public void setMaxMessageSize(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic,
- @ApiParam(value = "The max message size of the topic") int maxMessageSize) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalSetMaxMessageSize(maxMessageSize).whenComplete((r, ex) -> {
- if (ex instanceof RestException) {
- log.error("Failed updated persistence policies", ex);
- asyncResponse.resume(ex);
- } else if (ex != null) {
- log.error("Failed updated persistence policies", ex);
- asyncResponse.resume(new RestException(ex));
- } else {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
+ @ApiParam(value = "The max message size of the topic") int maxMessageSize) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalSetMaxMessageSize(maxMessageSize))
+ .thenRun(() -> {
log.info("[{}] Successfully set max message size: namespace={}, topic={}, maxMessageSiz={}",
clientAppId(),
namespaceName,
topicName.getLocalName(),
maxMessageSize);
asyncResponse.resume(Response.noContent().build());
- }
- });
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("setMaxMessageSize", ex, asyncResponse);
+ return null;
+ });
}
@DELETE
@@ -2118,23 +2272,25 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 405, message = "Topic level policy is disabled, to enable the topic level policy and retry"),
@ApiResponse(code = 409, message = "Concurrent modification")})
public void removeMaxMessageSize(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalSetMaxMessageSize(null).whenComplete((r, ex) -> {
- if (ex != null) {
- log.error("Failed to remove maxMessageSize", ex);
- asyncResponse.resume(new RestException(ex));
- } else {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalSetMaxMessageSize(null))
+ .thenRun(() -> {
log.info("[{}] Successfully remove max message size: namespace={}, topic={}",
clientAppId(),
namespaceName,
topicName.getLocalName());
asyncResponse.resume(Response.noContent().build());
- }
- });
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("removeMaxMessageSize", ex, asyncResponse);
+ return null;
+ });
}
@@ -2350,21 +2506,23 @@ public class PersistentTopics extends PersistentTopicsBase {
public void getDispatchRate(@Suspended final AsyncResponse asyncResponse,
@PathParam("tenant") String tenant,
@PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- try {
- Optional<DispatchRate> dispatchRate = internalGetDispatchRate();
- if (!dispatchRate.isPresent()) {
- asyncResponse.resume(Response.noContent().build());
- } else {
- asyncResponse.resume(dispatchRate.get());
- }
- } catch (RestException e) {
- asyncResponse.resume(e);
- } catch (Exception e) {
- asyncResponse.resume(new RestException(e));
- }
+ preValidation(authoritative)
+ .thenRun(() -> {
+ Optional<DispatchRate> dispatchRate = internalGetDispatchRate();
+ if (!dispatchRate.isPresent()) {
+ asyncResponse.resume(Response.noContent().build());
+ } else {
+ asyncResponse.resume(dispatchRate.get());
+ }
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("getDispatchRate", ex, asyncResponse);
+ return null;
+ });
}
@POST
@@ -2375,31 +2533,31 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 405, message = "Topic level policy is disabled, please enable the topic level policy and retry"),
@ApiResponse(code = 409, message = "Concurrent modification")})
public void setDispatchRate(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic,
- @ApiParam(value = "Dispatch rate for the specified topic") DispatchRate dispatchRate) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalSetDispatchRate(dispatchRate).whenComplete((r, ex) -> {
- if (ex instanceof RestException) {
- log.error("Failed to set topic dispatch rate", ex);
- asyncResponse.resume(ex);
- } else if (ex != null) {
- log.error("Failed to set topic dispatch rate");
- asyncResponse.resume(new RestException(ex));
- } else {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
+ @ApiParam(value = "Dispatch rate for the specified topic") DispatchRate dispatchRate) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalSetDispatchRate(dispatchRate))
+ .thenRun(() -> {
try {
- log.info("[{}] Successfully set topic dispatch rate: tenant={}, namespace={}, topic={}, dispatchRate={}",
- clientAppId(),
- tenant,
- namespace,
- topicName.getLocalName(),
- jsonMapper().writeValueAsString(dispatchRate));
+ log.info("[{}] Successfully set topic dispatch rate:"
+ + " tenant={}, namespace={}, topic={}, dispatchRate={}",
+ clientAppId(),
+ tenant,
+ namespace,
+ topicName.getLocalName(),
+ jsonMapper().writeValueAsString(dispatchRate));
} catch (JsonProcessingException ignore) {}
asyncResponse.resume(Response.noContent().build());
- }
- });
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("setDispatchRate", ex, asyncResponse);
+ return null;
+ });
}
@DELETE
@@ -2410,24 +2568,26 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 405, message = "Topic level policy is disabled, please enable the topic level policy and retry"),
@ApiResponse(code = 409, message = "Concurrent modification")})
public void removeDispatchRate(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalRemoveDispatchRate().whenComplete((r, ex) -> {
- if (ex != null) {
- log.error("Failed to remove topic dispatch rate", ex);
- asyncResponse.resume(new RestException(ex));
- } else {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalRemoveDispatchRate())
+ .thenRun(() -> {
log.info("[{}] Successfully remove topic dispatch rate: tenant={}, namespace={}, topic={}",
- clientAppId(),
- tenant,
- namespace,
- topicName.getLocalName());
+ clientAppId(),
+ tenant,
+ namespace,
+ topicName.getLocalName());
asyncResponse.resume(Response.noContent().build());
- }
- });
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("removeDispatchRate", ex, asyncResponse);
+ return null;
+ });
}
@GET
@@ -2440,21 +2600,23 @@ public class PersistentTopics extends PersistentTopicsBase {
public void getSubscriptionDispatchRate(@Suspended final AsyncResponse asyncResponse,
@PathParam("tenant") String tenant,
@PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- try {
- Optional<DispatchRate> dispatchRate = internalGetSubscriptionDispatchRate();
- if (!dispatchRate.isPresent()) {
- asyncResponse.resume(Response.noContent().build());
- } else {
- asyncResponse.resume(dispatchRate.get());
- }
- } catch (RestException e) {
- asyncResponse.resume(e);
- } catch (Exception e) {
- asyncResponse.resume(new RestException(e));
- }
+ preValidation(authoritative)
+ .thenRun(() -> {
+ Optional<DispatchRate> dispatchRate = internalGetSubscriptionDispatchRate();
+ if (!dispatchRate.isPresent()) {
+ asyncResponse.resume(Response.noContent().build());
+ } else {
+ asyncResponse.resume(dispatchRate.get());
+ }
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("getSubscriptionDispatchRate", ex, asyncResponse);
+ return null;
+ });
}
@POST
@@ -2465,31 +2627,31 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 405, message = "Topic level policy is disabled, please enable the topic level policy and retry"),
@ApiResponse(code = 409, message = "Concurrent modification")})
public void setSubscriptionDispatchRate(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic,
- @ApiParam(value = "Subscription message dispatch rate for the specified topic") DispatchRate dispatchRate) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalSetSubscriptionDispatchRate(dispatchRate).whenComplete((r, ex) -> {
- if (ex instanceof RestException) {
- log.error("Failed to set topic: {} subscription dispatch rate", topicName.getLocalName(), ex);
- asyncResponse.resume(ex);
- } else if (ex != null) {
- log.error("Failed to set topic: {} subscription dispatch rate", topicName.getLocalName());
- asyncResponse.resume(new RestException(ex));
- } else {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
+ @ApiParam(value = "Subscription message dispatch rate for the specified topic") DispatchRate dispatchRate) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalSetSubscriptionDispatchRate(dispatchRate))
+ .thenRun(() -> {
try {
- log.info("[{}] Successfully set topic subscription dispatch rate: tenant={}, namespace={}, topic={}, dispatchRate={}",
- clientAppId(),
- tenant,
- namespace,
- topicName.getLocalName(),
- jsonMapper().writeValueAsString(dispatchRate));
+ log.info("[{}] Successfully set topic subscription dispatch rate:"
+ + " tenant={}, namespace={}, topic={}, dispatchRate={}",
+ clientAppId(),
+ tenant,
+ namespace,
+ topicName.getLocalName(),
+ jsonMapper().writeValueAsString(dispatchRate));
} catch (JsonProcessingException ignore) {}
asyncResponse.resume(Response.noContent().build());
- }
- });
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("setSubscriptionDispatchRate", ex, asyncResponse);
+ return null;
+ });
}
@DELETE
@@ -2500,24 +2662,26 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 405, message = "Topic level policy is disabled, please enable the topic level policy and retry"),
@ApiResponse(code = 409, message = "Concurrent modification")})
public void removeSubscriptionDispatchRate(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalRemoveSubscriptionDispatchRate().whenComplete((r, ex) -> {
- if (ex != null) {
- log.error("Failed to remove topic: {} subscription dispatch rate", topicName.getLocalName(), ex);
- asyncResponse.resume(new RestException(ex));
- } else {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalRemoveSubscriptionDispatchRate())
+ .thenRun(() -> {
log.info("[{}] Successfully remove topic subscription dispatch rate: tenant={}, namespace={}, topic={}",
- clientAppId(),
- tenant,
- namespace,
- topicName.getLocalName());
+ clientAppId(),
+ tenant,
+ namespace,
+ topicName.getLocalName());
asyncResponse.resume(Response.noContent().build());
- }
- });
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("removeSubscriptionDispatchRate", ex, asyncResponse);
+ return null;
+ });
}
@GET
@@ -2528,23 +2692,25 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 405, message = "Topic level policy is disabled, please enable the topic level policy and retry"),
@ApiResponse(code = 409, message = "Concurrent modification")})
public void getCompactionThreshold(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- try {
- Optional<Long> compactionThreshold = internalGetCompactionThreshold();
- if (!compactionThreshold.isPresent()) {
- asyncResponse.resume(Response.noContent().build());
- } else {
- asyncResponse.resume(compactionThreshold.get());
- }
- } catch (RestException e) {
- asyncResponse.resume(e);
- } catch (Exception e) {
- asyncResponse.resume(new RestException(e));
- }
+ preValidation(authoritative)
+ .thenRun(() -> {
+ Optional<Long> compactionThreshold = internalGetCompactionThreshold();
+ if (!compactionThreshold.isPresent()) {
+ asyncResponse.resume(Response.noContent().build());
+ } else {
+ asyncResponse.resume(compactionThreshold.get());
+ }
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("getCompactionThreshold", ex, asyncResponse);
+ return null;
+ });
}
@POST
@@ -2555,31 +2721,31 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 405, message = "Topic level policy is disabled, please enable the topic level policy and retry"),
@ApiResponse(code = 409, message = "Concurrent modification")})
public void setCompactionThreshold(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic,
- @ApiParam(value = "Dispatch rate for the specified topic") long compactionThreshold) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalSetCompactionThreshold(compactionThreshold).whenComplete((r, ex) -> {
- if (ex instanceof RestException) {
- log.error("Failed to set topic dispatch rate", ex);
- asyncResponse.resume(ex);
- } else if (ex != null) {
- log.error("Failed to set topic dispatch rate");
- asyncResponse.resume(new RestException(ex));
- } else {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
+ @ApiParam(value = "Dispatch rate for the specified topic") long compactionThreshold) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalSetCompactionThreshold(compactionThreshold))
+ .thenRun(() -> {
try {
- log.info("[{}] Successfully set topic compaction threshold: tenant={}, namespace={}, topic={}, compactionThreshold={}",
- clientAppId(),
- tenant,
- namespace,
- topicName.getLocalName(),
- jsonMapper().writeValueAsString(compactionThreshold));
+ log.info("[{}] Successfully set topic compaction threshold:"
+ + " tenant={}, namespace={}, topic={}, compactionThreshold={}",
+ clientAppId(),
+ tenant,
+ namespace,
+ topicName.getLocalName(),
+ jsonMapper().writeValueAsString(compactionThreshold));
} catch (JsonProcessingException ignore) {}
asyncResponse.resume(Response.noContent().build());
- }
- });
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("setCompactionThreshold", ex, asyncResponse);
+ return null;
+ });
}
@DELETE
@@ -2590,24 +2756,26 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 405, message = "Topic level policy is disabled, please enable the topic level policy and retry"),
@ApiResponse(code = 409, message = "Concurrent modification")})
public void removeCompactionThreshold(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalRemoveCompactionThreshold().whenComplete((r, ex) -> {
- if (ex != null) {
- log.error("Failed to remove topic dispatch rate", ex);
- asyncResponse.resume(new RestException(ex));
- } else {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalRemoveCompactionThreshold())
+ .thenRun(() -> {
log.info("[{}] Successfully remove topic compaction threshold: tenant={}, namespace={}, topic={}",
- clientAppId(),
- tenant,
- namespace,
- topicName.getLocalName());
+ clientAppId(),
+ tenant,
+ namespace,
+ topicName.getLocalName());
asyncResponse.resume(Response.noContent().build());
- }
- });
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("removeCompactionThreshold", ex, asyncResponse);
+ return null;
+ });
}
@GET
@@ -2618,23 +2786,25 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 405, message = "Topic level policy is disabled, please enable the topic level policy and retry"),
@ApiResponse(code = 409, message = "Concurrent modification")})
public void getMaxConsumersPerSubscription(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- try {
- Optional<Integer> maxConsumersPerSubscription = internalGetMaxConsumersPerSubscription();
- if (!maxConsumersPerSubscription.isPresent()) {
- asyncResponse.resume(Response.noContent().build());
- } else {
- asyncResponse.resume(maxConsumersPerSubscription.get());
- }
- } catch (RestException e) {
- asyncResponse.resume(e);
- } catch (Exception e) {
- asyncResponse.resume(new RestException(e));
- }
+ preValidation(authoritative)
+ .thenRun(() -> {
+ Optional<Integer> maxConsumersPerSubscription = internalGetMaxConsumersPerSubscription();
+ if (!maxConsumersPerSubscription.isPresent()) {
+ asyncResponse.resume(Response.noContent().build());
+ } else {
+ asyncResponse.resume(maxConsumersPerSubscription.get());
+ }
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("getMaxConsumersPerSubscription", ex, asyncResponse);
+ return null;
+ });
}
@POST
@@ -2645,22 +2815,19 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 405, message = "Topic level policy is disabled, please enable the topic level policy and retry"),
@ApiResponse(code = 409, message = "Concurrent modification")})
public void setMaxConsumersPerSubscription(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic,
- @ApiParam(value = "Dispatch rate for the specified topic") int maxConsumersPerSubscription) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalSetMaxConsumersPerSubscription(maxConsumersPerSubscription).whenComplete((r, ex) -> {
- if (ex instanceof RestException) {
- log.error("Failed to set topic {} max consumers per subscription ", topicName.getLocalName(), ex);
- asyncResponse.resume(ex);
- } else if (ex != null) {
- log.error("Failed to set topic max consumers per subscription");
- asyncResponse.resume(new RestException(ex));
- } else {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
+ @ApiParam(value = "Dispatch rate for the specified topic") int maxConsumersPerSubscription) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalSetMaxConsumersPerSubscription(maxConsumersPerSubscription))
+ .thenRun(() -> {
try {
- log.info("[{}] Successfully set topic max consumers per subscription: tenant={}, namespace={}, topic={}, maxConsumersPerSubscription={}",
+ log.info("[{}] Successfully set topic max consumers per subscription:"
+ + " tenant={}, namespace={}, topic={}, maxConsumersPerSubscription={}",
clientAppId(),
tenant,
namespace,
@@ -2668,8 +2835,11 @@ public class PersistentTopics extends PersistentTopicsBase {
jsonMapper().writeValueAsString(maxConsumersPerSubscription));
} catch (JsonProcessingException ignore) {}
asyncResponse.resume(Response.noContent().build());
- }
- });
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("setMaxConsumersPerSubscription", ex, asyncResponse);
+ return null;
+ });
}
@DELETE
@@ -2680,24 +2850,27 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 405, message = "Topic level policy is disabled, please enable the topic level policy and retry"),
@ApiResponse(code = 409, message = "Concurrent modification")})
public void removeMaxConsumersPerSubscription(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalRemoveMaxConsumersPerSubscription().whenComplete((r, ex) -> {
- if (ex != null) {
- log.error("Failed to remove topic {} max consuners per subscription", topicName.getLocalName(), ex);
- asyncResponse.resume(new RestException(ex));
- } else {
- log.info("[{}] Successfully remove topic max consumers per subscription: tenant={}, namespace={}, topic={}",
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalRemoveMaxConsumersPerSubscription())
+ .thenRun(() -> {
+ log.info("[{}] Successfully remove topic max consumers per subscription:"
+ + " tenant={}, namespace={}, topic={}",
clientAppId(),
tenant,
namespace,
topicName.getLocalName());
asyncResponse.resume(Response.noContent().build());
- }
- });
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("removeMaxConsumersPerSubscription", ex, asyncResponse);
+ return null;
+ });
}
@GET
@@ -2710,21 +2883,23 @@ public class PersistentTopics extends PersistentTopicsBase {
public void getPublishRate(@Suspended final AsyncResponse asyncResponse,
@PathParam("tenant") String tenant,
@PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- try {
- Optional<PublishRate> publishRate = internalGetPublishRate();
- if (!publishRate.isPresent()) {
- asyncResponse.resume(Response.noContent().build());
- } else {
- asyncResponse.resume(publishRate.get());
- }
- } catch (RestException e) {
- asyncResponse.resume(e);
- } catch (Exception e) {
- asyncResponse.resume(new RestException(e));
- }
+ preValidation(authoritative)
+ .thenRun(() -> {
+ Optional<PublishRate> publishRate = internalGetPublishRate();
+ if (!publishRate.isPresent()) {
+ asyncResponse.resume(Response.noContent().build());
+ } else {
+ asyncResponse.resume(publishRate.get());
+ }
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("getPublishRate", ex, asyncResponse);
+ return null;
+ });
}
@POST
@@ -2735,31 +2910,31 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 405, message = "Topic level policy is disabled, please enable the topic level policy and retry"),
@ApiResponse(code = 409, message = "Concurrent modification")})
public void setPublishRate(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic,
- @ApiParam(value = "Dispatch rate for the specified topic") PublishRate publishRate) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalSetPublishRate(publishRate).whenComplete((r, ex) -> {
- if (ex instanceof RestException) {
- log.error("Failed to set topic dispatch rate", ex);
- asyncResponse.resume(ex);
- } else if (ex != null) {
- log.error("Failed to set topic dispatch rate");
- asyncResponse.resume(new RestException(ex));
- } else {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
+ @ApiParam(value = "Dispatch rate for the specified topic") PublishRate publishRate) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalSetPublishRate(publishRate))
+ .thenRun(() -> {
try {
- log.info("[{}] Successfully set topic publish rate: tenant={}, namespace={}, topic={}, publishRate={}",
- clientAppId(),
- tenant,
- namespace,
- topicName.getLocalName(),
- jsonMapper().writeValueAsString(publishRate));
+ log.info("[{}] Successfully set topic publish rate:"
+ + " tenant={}, namespace={}, topic={}, publishRate={}",
+ clientAppId(),
+ tenant,
+ namespace,
+ topicName.getLocalName(),
+ jsonMapper().writeValueAsString(publishRate));
} catch (JsonProcessingException ignore) {}
asyncResponse.resume(Response.noContent().build());
- }
- });
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("setPublishRate", ex, asyncResponse);
+ return null;
+ });
}
@DELETE
@@ -2770,24 +2945,26 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 405, message = "Topic level policy is disabled, please enable the topic level policy and retry"),
@ApiResponse(code = 409, message = "Concurrent modification")})
public void removePublishRate(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalRemovePublishRate().whenComplete((r, ex) -> {
- if (ex != null) {
- log.error("Failed to remove topic publish rate", ex);
- asyncResponse.resume(new RestException(ex));
- } else {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalRemovePublishRate())
+ .thenRun(() -> {
log.info("[{}] Successfully remove topic publish rate: tenant={}, namespace={}, topic={}",
- clientAppId(),
- tenant,
- namespace,
- topicName.getLocalName());
+ clientAppId(),
+ tenant,
+ namespace,
+ topicName.getLocalName());
asyncResponse.resume(Response.noContent().build());
- }
- });
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("removePublishRate", ex, asyncResponse);
+ return null;
+ });
}
@GET
@@ -2798,23 +2975,25 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 405, message = "Topic level policy is disabled, please enable the topic level policy and retry"),
@ApiResponse(code = 409, message = "Concurrent modification")})
public void getSubscribeRate(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- try {
- Optional<SubscribeRate> subscribeRate = internalGetSubscribeRate();
- if (!subscribeRate.isPresent()) {
- asyncResponse.resume(Response.noContent().build());
- } else {
- asyncResponse.resume(subscribeRate.get());
- }
- } catch (RestException e) {
- asyncResponse.resume(e);
- } catch (Exception e) {
- asyncResponse.resume(new RestException(e));
- }
+ preValidation(authoritative)
+ .thenRun(() -> {
+ Optional<SubscribeRate> subscribeRate = internalGetSubscribeRate();
+ if (!subscribeRate.isPresent()) {
+ asyncResponse.resume(Response.noContent().build());
+ } else {
+ asyncResponse.resume(subscribeRate.get());
+ }
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("getSubscribeRate", ex, asyncResponse);
+ return null;
+ });
}
@POST
@@ -2825,22 +3004,19 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 405, message = "Topic level policy is disabled, please enable the topic level policy and retry"),
@ApiResponse(code = 409, message = "Concurrent modification")})
public void setSubscribeRate(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic,
- @ApiParam(value = "Subscribe rate for the specified topic") SubscribeRate subscribeRate) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalSetSubscribeRate(subscribeRate).whenComplete((r, ex) -> {
- if (ex instanceof RestException) {
- log.error("Failed to set topic {} subscribe rate", topicName.getLocalName(), ex);
- asyncResponse.resume(ex);
- } else if (ex != null) {
- log.error("Failed to set topic subscribe rate");
- asyncResponse.resume(new RestException(ex));
- } else {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
+ @ApiParam(value = "Subscribe rate for the specified topic") SubscribeRate subscribeRate) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalSetSubscribeRate(subscribeRate))
+ .thenRun(() -> {
try {
- log.info("[{}] Successfully set topic subscribe rate: tenant={}, namespace={}, topic={}, subscribeRate={}",
+ log.info("[{}] Successfully set topic subscribe rate:"
+ + " tenant={}, namespace={}, topic={}, subscribeRate={}",
clientAppId(),
tenant,
namespace,
@@ -2848,8 +3024,11 @@ public class PersistentTopics extends PersistentTopicsBase {
jsonMapper().writeValueAsString(subscribeRate));
} catch (JsonProcessingException ignore) {}
asyncResponse.resume(Response.noContent().build());
- }
- });
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("setSubscribeRate", ex, asyncResponse);
+ return null;
+ });
}
@DELETE
@@ -2860,24 +3039,26 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 405, message = "Topic level policy is disabled, please enable the topic level policy and retry"),
@ApiResponse(code = 409, message = "Concurrent modification")})
public void removeSubscribeRate(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace,
- @PathParam("topic") @Encoded String encodedTopic) {
- validateTopicName(tenant, namespace, encodedTopic);
- preValidation();
- internalRemoveSubscribeRate().whenComplete((r, ex) -> {
- if (ex != null) {
- log.error("Failed to remove topic {} subscribe rate ", topicName.getLocalName(), ex);
- asyncResponse.resume(new RestException(ex));
- } else {
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace,
+ @PathParam("topic") @Encoded String encodedTopic,
+ @ApiParam(value = "Is authentication required to perform this operation")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
+ validateTopicName(tenant, namespace, encodedTopic);
+ preValidation(authoritative)
+ .thenCompose(__ -> internalRemoveSubscribeRate())
+ .thenRun(() -> {
log.info("[{}] Successfully remove topic subscribe rate: tenant={}, namespace={}, topic={}",
clientAppId(),
tenant,
namespace,
topicName.getLocalName());
asyncResponse.resume(Response.noContent().build());
- }
- });
+ })
+ .exceptionally(ex -> {
+ handleTopicPolicyException("removeSubscribeRate", ex, asyncResponse);
+ return null;
+ });
}
private static final Logger log = LoggerFactory.getLogger(PersistentTopics.class);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
index 580ca0d..52cce38 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
@@ -234,7 +234,7 @@ public class NamespaceService {
return bundleFactory.getFullBundle(fqnn);
}
- private CompletableFuture<NamespaceBundle> getFullBundleAsync(NamespaceName fqnn) throws Exception {
+ private CompletableFuture<NamespaceBundle> getFullBundleAsync(NamespaceName fqnn) {
return bundleFactory.getFullBundleAsync(fqnn);
}
@@ -243,7 +243,7 @@ public class NamespaceService {
*
* If the service unit is not owned, return a CompletableFuture with empty optional.
*/
- public CompletableFuture<Optional<URL>> getWebServiceUrlAsync(ServiceUnitId suName, LookupOptions options) throws Exception {
+ public CompletableFuture<Optional<URL>> getWebServiceUrlAsync(ServiceUnitId suName, LookupOptions options) {
if (suName instanceof TopicName) {
TopicName name = (TopicName) suName;
if (LOG.isDebugEnabled()) {
@@ -945,6 +945,24 @@ public class NamespaceService {
throw new IllegalArgumentException("Invalid class of NamespaceBundle: " + suName.getClass().getName());
}
+ public CompletableFuture<Boolean> isServiceUnitOwnedAsync(ServiceUnitId suName) {
+ if (suName instanceof TopicName) {
+ return isTopicOwnedAsync((TopicName) suName);
+ }
+
+ if (suName instanceof NamespaceName) {
+ return isNamespaceOwnedAsync((NamespaceName) suName);
+ }
+
+ if (suName instanceof NamespaceBundle) {
+ return CompletableFuture.completedFuture(
+ ownershipCache.isNamespaceBundleOwned((NamespaceBundle) suName));
+ }
+
+ return FutureUtil.failedFuture(
+ new IllegalArgumentException("Invalid class of NamespaceBundle: " + suName.getClass().getName()));
+ }
+
public boolean isServiceUnitActive(TopicName topicName) {
try {
return ownershipCache.getOwnedBundle(getBundle(topicName)).isActive();
@@ -958,6 +976,11 @@ public class NamespaceService {
return ownershipCache.getOwnedBundle(getFullBundle(fqnn)) != null;
}
+ private CompletableFuture<Boolean> isNamespaceOwnedAsync(NamespaceName fqnn) {
+ return getFullBundleAsync(fqnn)
+ .thenApply(bundle -> ownershipCache.getOwnedBundle(bundle) != null);
+ }
+
private CompletableFuture<Boolean> isTopicOwnedAsync(TopicName topic) {
return getBundleAsync(topic).thenApply(bundle -> ownershipCache.isNamespaceBundleOwned(bundle));
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsGenerator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsGenerator.java
index 7853503..80bc06f 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsGenerator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsGenerator.java
@@ -101,8 +101,11 @@ public class PrometheusMetricsGenerator {
NamespaceStatsAggregator.generate(pulsar, includeTopicMetrics, includeConsumerMetrics, stream);
- FunctionsStatsGenerator.generate(pulsar.getWorkerService(),
- pulsar.getConfiguration().getClusterName(), stream);
+ try {
+ FunctionsStatsGenerator.generate(pulsar.getWorkerService(),
+ pulsar.getConfiguration().getClusterName(), stream);
+ } catch (UnsupportedOperationException ignore) {
+ }
generateBrokerBasicMetrics(pulsar, stream);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
index 74176a4..1130810 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
@@ -47,6 +47,7 @@ import com.google.common.base.Splitter;
import com.google.common.collect.BoundType;
import com.google.common.collect.Range;
import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.admin.AdminResource;
@@ -481,6 +482,60 @@ public abstract class PulsarWebResource {
}
}
+ protected CompletableFuture<Void> validateTopicOwnershipAsync(TopicName topicName, boolean authoritative) {
+ NamespaceService nsService = pulsar().getNamespaceService();
+
+ LookupOptions options = LookupOptions.builder()
+ .authoritative(authoritative)
+ .requestHttps(isRequestHttps())
+ .readOnly(false)
+ .loadTopicsInBundle(false)
+ .build();
+
+ return nsService.getWebServiceUrlAsync(topicName, options)
+ .thenApply(webUrl -> {
+ // Ensure we get a url
+ if (webUrl == null || !webUrl.isPresent()) {
+ log.info("Unable to get web service url");
+ throw new RestException(Status.PRECONDITION_FAILED,
+ "Failed to find ownership for topic:" + topicName);
+ }
+ return webUrl.get();
+ }).thenCompose(webUrl -> nsService.isServiceUnitOwnedAsync(topicName)
+ .thenApply(isTopicOwned -> Pair.of(webUrl, isTopicOwned))
+ ).thenAccept(pair -> {
+ URL webUrl = pair.getLeft();
+ boolean isTopicOwned = pair.getRight();
+
+ if (!isTopicOwned) {
+ boolean newAuthoritative = isLeaderBroker(pulsar());
+ // Replace the host and port of the current request and redirect
+ URI redirect = UriBuilder.fromUri(uri.getRequestUri())
+ .host(webUrl.getHost())
+ .port(webUrl.getPort())
+ .replaceQueryParam("authoritative", newAuthoritative)
+ .build();
+ // Redirect
+ if (log.isDebugEnabled()) {
+ log.debug("Redirecting the rest call to {}", redirect);
+ }
+ throw new WebApplicationException(Response.temporaryRedirect(redirect).build());
+ }
+ }).exceptionally(ex -> {
+ if (ex.getCause() instanceof IllegalArgumentException
+ || ex.getCause() instanceof IllegalStateException) {
+ if (log.isDebugEnabled()) {
+ log.debug("Failed to find owner for topic: {}", topicName, ex);
+ }
+ throw new RestException(Status.PRECONDITION_FAILED, "Can't find owner for topic " + topicName);
+ } else if (ex.getCause() instanceof WebApplicationException) {
+ throw (WebApplicationException) ex.getCause();
+ } else {
+ throw new RestException(ex.getCause());
+ }
+ });
+ }
+
protected NamespaceBundle validateNamespaceBundleRange(NamespaceName fqnn, BundlesData bundles,
String bundleRange) {
try {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundleFactory.java b/pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundleFactory.java
index 0e0b031..e723475 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundleFactory.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundleFactory.java
@@ -181,7 +181,7 @@ public class NamespaceBundleFactory implements ZooKeeperCacheListener<LocalPolic
return bundlesCache.synchronous().get(fqnn).getFullBundle();
}
- public CompletableFuture<NamespaceBundle> getFullBundleAsync(NamespaceName fqnn) throws Exception {
+ public CompletableFuture<NamespaceBundle> getFullBundleAsync(NamespaceName fqnn) {
return bundlesCache.get(fqnn).thenApply(NamespaceBundles::getFullBundle);
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java
index 1e2b2d1..3b778ee 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java
@@ -279,8 +279,6 @@ public class AdminApiTest2 extends MockedPulsarServiceBaseTest {
int partitionNum = 3;
admin.topics().createPartitionedTopic(topic, partitionNum);
pulsarClient.newConsumer().topic(topic).subscriptionName("sub").subscribe().close();
- TopicName topicName = TopicName.get(topic);
- Awaitility.await().until(()-> pulsar.getTopicPoliciesService().cacheIsInitialized(topicName));
setTopicPoliciesAndValidate(admin2, admin3, topic);
//for non-partitioned topic, we can get topic policies from every broker
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java
index 15cd6bc..1613911 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java
@@ -367,23 +367,32 @@ public class TopicPoliciesTest extends MockedPulsarServiceBaseTest {
Awaitility.await().atMost(3, TimeUnit.SECONDS)
.until(() -> pulsar.getTopicPoliciesService().cacheIsInitialized(TopicName.get(testTopic)));
+ admin.topics().createNonPartitionedTopic(persistenceTopic);
+
admin.topics().setPersistence(persistenceTopic, persistencePolicies);
Awaitility.await().atMost(3, TimeUnit.SECONDS)
.untilAsserted(() -> Assert.assertEquals(admin.topics().getPersistence(persistenceTopic), persistencePolicies));
- admin.lookups().lookupTopic(persistenceTopic);
+ Consumer<byte[]> consumer = pulsarClient.newConsumer()
+ .topic(persistenceTopic)
+ .subscriptionName("test")
+ .subscribe();
+ admin.topics().unload(persistenceTopic);
Topic t = pulsar.getBrokerService().getOrCreateTopic(persistenceTopic).get();
PersistentTopic persistentTopic = (PersistentTopic) t;
- ManagedLedgerConfig managedLedgerConfig = persistentTopic.getManagedLedger().getConfig();
- assertEquals(managedLedgerConfig.getEnsembleSize(), 3);
- assertEquals(managedLedgerConfig.getWriteQuorumSize(), 3);
- assertEquals(managedLedgerConfig.getAckQuorumSize(), 3);
- assertEquals(managedLedgerConfig.getThrottleMarkDelete(), 0.1);
+ Awaitility.await().untilAsserted(() -> {
+ ManagedLedgerConfig managedLedgerConfig = persistentTopic.getManagedLedger().getConfig();
+ assertEquals(managedLedgerConfig.getEnsembleSize(), 3);
+ assertEquals(managedLedgerConfig.getWriteQuorumSize(), 3);
+ assertEquals(managedLedgerConfig.getAckQuorumSize(), 3);
+ assertEquals(managedLedgerConfig.getThrottleMarkDelete(), 0.1);
+ });
PersistencePolicies getPersistencePolicies = admin.topics().getPersistence(persistenceTopic);
log.info("PersistencePolicies: {} will set to the topic: {}", persistencePolicies, persistenceTopic);
Assert.assertEquals(getPersistencePolicies, persistencePolicies);
+ consumer.close();
}
@Test
@@ -400,6 +409,7 @@ public class TopicPoliciesTest extends MockedPulsarServiceBaseTest {
Awaitility.await().atMost(3, TimeUnit.SECONDS)
.until(() -> pulsar.getTopicPoliciesService().cacheIsInitialized(TopicName.get(testTopic)));
+ admin.topics().createNonPartitionedTopic(persistenceTopic);
admin.topics().setPersistence(persistenceTopic, persistencePolicies);
Awaitility.await().atMost(3, TimeUnit.SECONDS)
@@ -442,12 +452,12 @@ public class TopicPoliciesTest extends MockedPulsarServiceBaseTest {
Awaitility.await().atMost(3, TimeUnit.SECONDS)
.until(() -> pulsar.getTopicPoliciesService().cacheIsInitialized(TopicName.get(testTopic)));
+ admin.topics().createPartitionedTopic(persistenceTopic, 2);
admin.topics().setMaxProducers(persistenceTopic, maxProducers);
Awaitility.await().atMost(3, TimeUnit.SECONDS)
.untilAsserted(() -> Assert.assertEquals(admin.topics().getMaxProducers(persistenceTopic), maxProducers));
- admin.topics().createPartitionedTopic(persistenceTopic, 2);
Producer<byte[]> producer1 = pulsarClient.newProducer().topic(persistenceTopic).create();
Producer<byte[]> producer2 = pulsarClient.newProducer().topic(persistenceTopic).create();
Producer<byte[]> producer3 = null;
@@ -474,12 +484,12 @@ public class TopicPoliciesTest extends MockedPulsarServiceBaseTest {
Awaitility.await().atMost(3, TimeUnit.SECONDS)
.until(() -> pulsar.getTopicPoliciesService().cacheIsInitialized(TopicName.get(testTopic)));
+ admin.topics().createPartitionedTopic(persistenceTopic, 2);
admin.topics().setMaxProducers(persistenceTopic, maxProducers);
Awaitility.await().atMost(3, TimeUnit.SECONDS)
.untilAsserted(() -> Assert.assertEquals(admin.topics().getMaxProducers(persistenceTopic), maxProducers));
- admin.topics().createPartitionedTopic(persistenceTopic, 2);
Producer<byte[]> producer1 = pulsarClient.newProducer().topic(persistenceTopic).create();
Producer<byte[]> producer2 = pulsarClient.newProducer().topic(persistenceTopic).create();
Producer<byte[]> producer3 = null;
@@ -968,12 +978,12 @@ public class TopicPoliciesTest extends MockedPulsarServiceBaseTest {
Awaitility.await().atMost(3, TimeUnit.SECONDS)
.until(() -> pulsar.getTopicPoliciesService().cacheIsInitialized(TopicName.get(testTopic)));
+ admin.topics().createPartitionedTopic(persistenceTopic, 2);
admin.topics().setMaxConsumers(persistenceTopic, maxConsumers);
Awaitility.await().atMost(3, TimeUnit.SECONDS)
.untilAsserted(() -> Assert.assertEquals(admin.topics().getMaxConsumers(persistenceTopic), maxConsumers));
- admin.topics().createPartitionedTopic(persistenceTopic, 2);
Consumer<byte[]> consumer1 = pulsarClient.newConsumer().subscriptionName("sub1").topic(persistenceTopic).subscribe();
Consumer<byte[]> consumer2 = pulsarClient.newConsumer().subscriptionName("sub2").topic(persistenceTopic).subscribe();
Consumer<byte[]> consumer3 = null;
@@ -1000,12 +1010,12 @@ public class TopicPoliciesTest extends MockedPulsarServiceBaseTest {
Awaitility.await().atMost(3, TimeUnit.SECONDS)
.until(() -> pulsar.getTopicPoliciesService().cacheIsInitialized(TopicName.get(testTopic)));
+ admin.topics().createPartitionedTopic(persistenceTopic, 2);
admin.topics().setMaxConsumers(persistenceTopic, maxConsumers);
Awaitility.await().atMost(3, TimeUnit.SECONDS)
.untilAsserted(() -> Assert.assertEquals(admin.topics().getMaxConsumers(persistenceTopic), maxConsumers));
- admin.topics().createPartitionedTopic(persistenceTopic, 2);
Consumer<byte[]> consumer1 = pulsarClient.newConsumer().subscriptionName("sub1").topic(persistenceTopic).subscribe();
Consumer<byte[]> consumer2 = pulsarClient.newConsumer().subscriptionName("sub2").topic(persistenceTopic).subscribe();
Consumer<byte[]> consumer3 = null;
@@ -1257,6 +1267,7 @@ public class TopicPoliciesTest extends MockedPulsarServiceBaseTest {
public void testTopicMaxMessageSizeApi() throws Exception{
Awaitility.await().atMost(5, TimeUnit.SECONDS)
.until(() -> pulsar.getTopicPoliciesService().cacheIsInitialized(TopicName.get(persistenceTopic)));
+ admin.topics().createNonPartitionedTopic(persistenceTopic);
assertNull(admin.topics().getMaxMessageSize(persistenceTopic));
admin.topics().setMaxMessageSize(persistenceTopic,10);
@@ -1341,7 +1352,7 @@ public class TopicPoliciesTest extends MockedPulsarServiceBaseTest {
Awaitility.await().atMost(5, TimeUnit.SECONDS)
.until(() -> pulsar.getTopicPoliciesService().cacheIsInitialized(TopicName.get(topic)));
//should not fail
- assertNull(admin.topics().getMessageTTL(topic));
+ assertEquals(admin.topics().getMessageTTL(topic), 0);
}
@Test(timeOut = 20000)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
index 0fca6ea..30390d3 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
@@ -996,12 +996,7 @@ public class BrokerServiceTest extends BrokerTestBase {
@Test
public void testMetricsProvider() throws IOException {
- PrometheusRawMetricsProvider rawMetricsProvider = new PrometheusRawMetricsProvider() {
- @Override
- public void generate(SimpleTextOutputStream stream) {
- stream.write("test_metrics{label1=\"xyz\"} 10 \n");
- }
- };
+ PrometheusRawMetricsProvider rawMetricsProvider = stream -> stream.write("test_metrics{label1=\"xyz\"} 10 \n");
getPulsar().addPrometheusRawMetricsProvider(rawMetricsProvider);
HttpClient httpClient = HttpClientBuilder.create().build();
final String metricsEndPoint = getPulsar().getWebServiceAddress() + "/metrics";
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java
index 4db81d3..69d1868 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java
@@ -3585,7 +3585,7 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
log.info("-- Exiting {} test --", methodName);
}
- @Test(timeOut = 5000)
+ @Test(timeOut = 10000)
public void testReceiveAsyncCompletedWhenClosing() throws Exception {
final String topic = "persistent://my-property/my-ns/testCompletedWhenClosing";
final String partitionedTopic = "persistent://my-property/my-ns/testCompletedWhenClosing-partitioned";
diff --git a/pulsar-discovery-service/src/test/java/org/apache/pulsar/discovery/service/DiscoveryServiceTest.java b/pulsar-discovery-service/src/test/java/org/apache/pulsar/discovery/service/DiscoveryServiceTest.java
index 4427ddd..f19ab7a 100644
--- a/pulsar-discovery-service/src/test/java/org/apache/pulsar/discovery/service/DiscoveryServiceTest.java
+++ b/pulsar-discovery-service/src/test/java/org/apache/pulsar/discovery/service/DiscoveryServiceTest.java
@@ -33,7 +33,6 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
-import org.apache.bookkeeper.bookie.BookieException;
import org.apache.bookkeeper.util.ZkUtils;
import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.common.api.proto.PulsarApi.BaseCommand;
@@ -140,7 +139,7 @@ public class DiscoveryServiceTest extends BaseDiscoveryTestSetup {
future.get();
fail("Partition metadata lookup should have failed");
} catch (ExecutionException e) {
- assertEquals(e.getCause().getClass(), BookieException.MetadataStoreException.class);
+ assertEquals(e.getCause().getClass(), SessionExpiredException.class);
}
}