codelipenghui commented on code in PR #20330:
URL: https://github.com/apache/pulsar/pull/20330#discussion_r1265036101
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Transactions.java:
##########
@@ -332,6 +334,51 @@ public void getPendingAckInternalStats(@Suspended final
AsyncResponse asyncRespo
}
}
+ @GET
+ @Path("/transactionBufferInternalStats/{tenant}/{namespace}/{topic}")
+ @ApiOperation(value = "Get transaction buffer internal stats.")
+ @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"),
+ @ApiResponse(code = 503, message = "This Broker is not enable
transaction"),
+ @ApiResponse(code = 307, message = "Topic is not owned by this
broker!"),
+ @ApiResponse(code = 405, message = "Transaction buffer don't use
managedLedger!"),
+ @ApiResponse(code = 400, message = "Topic is not a persistent
topic!"),
+ @ApiResponse(code = 409, message = "Concurrent modification")
+ })
+ public void getTransactionBufferInternalStats(@Suspended final
AsyncResponse asyncResponse,
+ @QueryParam("authoritative")
+ @DefaultValue("false")
boolean authoritative,
+ @PathParam("tenant") String
tenant,
+ @PathParam("namespace")
String namespace,
+ @PathParam("topic") @Encoded
String encodedTopic,
+ @QueryParam("metadata")
@DefaultValue("false") boolean metadata) {
+ try {
+ validateTopicName(tenant, namespace, encodedTopic);
+ internalGetTransactionBufferInternalStats(authoritative, metadata)
+ .thenAccept(asyncResponse::resume)
+ .exceptionally(ex -> {
+ if (!isRedirectException(ex)) {
+ log.error("[{}] Failed to get transaction buffer
internal stats {}",
+ clientAppId(), topicName, ex);
+ }
+ Throwable cause =
FutureUtil.unwrapCompletionException(ex);
+ if (cause instanceof
BrokerServiceException.ServiceUnitNotReadyException) {
+ asyncResponse.resume(new
RestException(SERVICE_UNAVAILABLE, cause));
+ } else if (cause instanceof
BrokerServiceException.NotAllowedException) {
+ asyncResponse.resume(new
RestException(METHOD_NOT_ALLOWED, cause));
+ } else if (cause instanceof
BrokerServiceException.SubscriptionNotFoundException) {
+ asyncResponse.resume(new RestException(NOT_FOUND,
cause));
+ } else {
+ asyncResponse.resume(new RestException(cause));
+ }
Review Comment:
We have a method named `resumeAsyncResponseExceptionally` in
PulsarWebResource.java. Can we use this method to avoid the duplicated code?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
