nodece commented on code in PR #15621:
URL: https://github.com/apache/pulsar/pull/15621#discussion_r880004483
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/Namespaces.java:
##########
@@ -572,14 +579,26 @@ public void setAutoTopicCreation(@Suspended final
AsyncResponse asyncResponse,
@PathParam("property") String property,
@PathParam("cluster") String cluster,
@PathParam("namespace") String namespace,
AutoTopicCreationOverride
autoTopicCreationOverride) {
- try {
- validateNamespaceName(property, cluster, namespace);
- internalSetAutoTopicCreation(asyncResponse,
autoTopicCreationOverride);
- } catch (RestException e) {
- asyncResponse.resume(e);
- } catch (Exception e) {
- asyncResponse.resume(new RestException(e));
- }
+ validateNamespaceName(property, cluster, namespace);
+ internalSetAutoTopicCreationAsync(autoTopicCreationOverride)
+ .thenAccept(__ -> {
+ String autoOverride = (autoTopicCreationOverride != null
+ &&
autoTopicCreationOverride.isAllowAutoTopicCreation()) ? "enabled" : "disabled";
+ log.info("[{}] Successfully {} autoTopicCreation on
namespace {}", clientAppId(),
+
autoOverride, namespaceName);
+ asyncResponse.resume(Response.noContent().build());
+ })
+ .exceptionally(e -> {
+ log.error("[{}] Failed to set autoTopicCreation status on
namespace {}", clientAppId(),
Review Comment:
Throwable throwable = FutureUtil.unwrapCompletionException(e);
```suggestion
log.error("[{}] Failed to set autoTopicCreation status
on namespace {}", clientAppId(), namespaceName, throwable)
```
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/Namespaces.java:
##########
@@ -572,14 +579,26 @@ public void setAutoTopicCreation(@Suspended final
AsyncResponse asyncResponse,
@PathParam("property") String property,
@PathParam("cluster") String cluster,
@PathParam("namespace") String namespace,
AutoTopicCreationOverride
autoTopicCreationOverride) {
- try {
- validateNamespaceName(property, cluster, namespace);
- internalSetAutoTopicCreation(asyncResponse,
autoTopicCreationOverride);
- } catch (RestException e) {
- asyncResponse.resume(e);
- } catch (Exception e) {
- asyncResponse.resume(new RestException(e));
- }
+ validateNamespaceName(property, cluster, namespace);
+ internalSetAutoTopicCreationAsync(autoTopicCreationOverride)
+ .thenAccept(__ -> {
+ String autoOverride = (autoTopicCreationOverride != null
+ &&
autoTopicCreationOverride.isAllowAutoTopicCreation()) ? "enabled" : "disabled";
+ log.info("[{}] Successfully {} autoTopicCreation on
namespace {}", clientAppId(),
+
autoOverride, namespaceName);
+ asyncResponse.resume(Response.noContent().build());
+ })
+ .exceptionally(e -> {
+ log.error("[{}] Failed to set autoTopicCreation status on
namespace {}", clientAppId(),
+ namespaceName,
+ e.getCause());
+ if (FutureUtil.unwrapCompletionException(e) instanceof
NotFoundException) {
Review Comment:
```suggestion
if (throwable instanceof NotFoundException) {
```
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java:
##########
@@ -521,14 +540,24 @@ public void setAutoTopicCreation(
@ApiResponse(code = 404, message = "Tenant or cluster or namespace
doesn't exist") })
public void removeAutoTopicCreation(@Suspended final AsyncResponse
asyncResponse,
@PathParam("tenant") String tenant,
@PathParam("namespace") String namespace) {
- try {
- validateNamespaceName(tenant, namespace);
- internalRemoveAutoTopicCreation(asyncResponse);
- } catch (RestException e) {
- asyncResponse.resume(e);
- } catch (Exception e) {
- asyncResponse.resume(new RestException(e));
- }
+ validateNamespaceName(tenant, namespace);
+ internalSetAutoTopicCreationAsync(null)
+ .thenAccept(__ -> {
+ log.info("[{}] Successfully remove autoTopicCreation on
namespace {}",
+ clientAppId(), namespaceName);
+ asyncResponse.resume(Response.noContent().build());
+ })
+ .exceptionally(e -> {
+ log.error("[{}] Failed to remove autoTopicCreation status
on namespace {}", clientAppId(),
Review Comment:
Same above.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/Namespaces.java:
##########
@@ -590,14 +609,24 @@ public void setAutoTopicCreation(@Suspended final
AsyncResponse asyncResponse,
public void removeAutoTopicCreation(@Suspended final AsyncResponse
asyncResponse,
@PathParam("property") String
property, @PathParam("cluster") String cluster,
@PathParam("namespace") String
namespace) {
- try {
validateNamespaceName(property, cluster, namespace);
- internalRemoveAutoTopicCreation(asyncResponse);
- } catch (RestException e) {
- asyncResponse.resume(e);
- } catch (Exception e) {
- asyncResponse.resume(new RestException(e));
- }
+ internalSetAutoTopicCreationAsync(null)
+ .thenAccept(__ -> {
+ log.info("[{}] Successfully remove autoTopicCreation
on namespace {}",
+ clientAppId(), namespaceName);
+ asyncResponse.resume(Response.noContent().build());
+ })
+ .exceptionally(e -> {
+ log.error("[{}] Failed to remove autoTopicCreation
status on namespace {}", clientAppId(),
Review Comment:
Same above.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java:
##########
@@ -819,56 +818,38 @@ protected void
internalSetSubscriptionExpirationTime(Integer expirationTime) {
});
}
- protected AutoTopicCreationOverride internalGetAutoTopicCreation() {
- validateNamespacePolicyOperation(namespaceName,
PolicyName.AUTO_TOPIC_CREATION, PolicyOperation.READ);
- Policies policies = getNamespacePolicies(namespaceName);
- return policies.autoTopicCreationOverride;
- }
-
- protected void internalSetAutoTopicCreation(AsyncResponse asyncResponse,
- AutoTopicCreationOverride
autoTopicCreationOverride) {
- final int maxPartitions =
pulsar().getConfig().getMaxNumPartitionsPerPartitionedTopic();
- validateNamespacePolicyOperation(namespaceName,
PolicyName.AUTO_TOPIC_CREATION, PolicyOperation.WRITE);
- validatePoliciesReadOnlyAccess();
- if (autoTopicCreationOverride != null) {
- ValidateResult validateResult =
AutoTopicCreationOverrideImpl.validateOverride(autoTopicCreationOverride);
- if (!validateResult.isSuccess()) {
- throw new RestException(Status.PRECONDITION_FAILED,
- "Invalid configuration for autoTopicCreationOverride.
the detail is "
- + validateResult.getErrorInfo());
- }
- if (Objects.equals(autoTopicCreationOverride.getTopicType(),
TopicType.PARTITIONED.toString())) {
- if (maxPartitions > 0 &&
autoTopicCreationOverride.getDefaultNumPartitions() > maxPartitions) {
- throw new RestException(Status.NOT_ACCEPTABLE,
- "Number of partitions should be less than or equal
to " + maxPartitions);
- }
- }
- }
- // Force to read the data s.t. the watch to the cache content is setup.
- namespaceResources().setPoliciesAsync(namespaceName, policies -> {
- policies.autoTopicCreationOverride = autoTopicCreationOverride;
- return policies;
- }).thenApply(r -> {
- String autoOverride = (autoTopicCreationOverride != null
- && autoTopicCreationOverride.isAllowAutoTopicCreation()) ?
"enabled" : "disabled";
- log.info("[{}] Successfully {} autoTopicCreation on namespace {}",
clientAppId(),
- autoOverride != null ? autoOverride : "removed",
namespaceName);
- asyncResponse.resume(Response.noContent().build());
- return null;
- }).exceptionally(e -> {
- log.error("[{}] Failed to modify autoTopicCreation status on
namespace {}", clientAppId(), namespaceName,
- e.getCause());
- if (e.getCause() instanceof NotFoundException) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND,
"Namespace does not exist"));
- return null;
- }
- asyncResponse.resume(new RestException(e.getCause()));
- return null;
- });
+ protected CompletableFuture<AutoTopicCreationOverride>
internalGetAutoTopicCreationAsync() {
+ return validateNamespacePolicyOperationAsync(namespaceName,
PolicyName.AUTO_TOPIC_CREATION,
+ PolicyOperation.READ)
+ .thenCompose(__ -> getNamespacePoliciesAsync(namespaceName))
+ .thenApply(policies -> policies.autoTopicCreationOverride);
}
- protected void internalRemoveAutoTopicCreation(AsyncResponse
asyncResponse) {
- internalSetAutoTopicCreation(asyncResponse, null);
+ protected CompletableFuture<Void> internalSetAutoTopicCreationAsync(
+ AutoTopicCreationOverride autoTopicCreationOverride) {
+ return validateNamespacePolicyOperationAsync(namespaceName,
+ PolicyName.AUTO_TOPIC_CREATION, PolicyOperation.WRITE)
+ .thenCompose(__ -> validatePoliciesReadOnlyAccessAsync())
+ .thenAccept(__ -> {
+ int maxPartitions =
pulsar().getConfig().getMaxNumPartitionsPerPartitionedTopic();
+ if (autoTopicCreationOverride != null) {
+ ValidateResult validateResult =
+
AutoTopicCreationOverrideImpl.validateOverride(autoTopicCreationOverride);
+ if (!validateResult.isSuccess()) {
+ throw new RestException(Status.PRECONDITION_FAILED,
+ "Invalid configuration for
autoTopicCreationOverride. the detail is "
+ + validateResult.getErrorInfo());
+ }
+ if (maxPartitions > 0 &&
autoTopicCreationOverride.getDefaultNumPartitions() > maxPartitions) {
Review Comment:
`if (Objects.equals(autoTopicCreationOverride.getTopicType(),
TopicType.PARTITIONED.toString()))` is miss.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java:
##########
@@ -504,14 +511,26 @@ public void setAutoTopicCreation(
@PathParam("tenant") String tenant, @PathParam("namespace") String
namespace,
@ApiParam(value = "Settings for automatic topic creation",
required = true)
AutoTopicCreationOverride autoTopicCreationOverride) {
- try {
- validateNamespaceName(tenant, namespace);
- internalSetAutoTopicCreation(asyncResponse,
autoTopicCreationOverride);
- } catch (RestException e) {
- asyncResponse.resume(e);
- } catch (Exception e) {
- asyncResponse.resume(new RestException(e));
- }
+ validateNamespaceName(tenant, namespace);
+ internalSetAutoTopicCreationAsync(autoTopicCreationOverride)
+ .thenAccept(__ -> {
+ String autoOverride = (autoTopicCreationOverride != null
+ &&
autoTopicCreationOverride.isAllowAutoTopicCreation()) ? "enabled" : "disabled";
+ log.info("[{}] Successfully {} autoTopicCreation on
namespace {}", clientAppId(),
+ autoOverride, namespaceName);
+ asyncResponse.resume(Response.noContent().build());
+ })
+ .exceptionally(e -> {
+ log.error("[{}] Failed to set autoTopicCreation status on
namespace {}", clientAppId(),
Review Comment:
Same above.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
