nodece commented on code in PR #15694:
URL: https://github.com/apache/pulsar/pull/15694#discussion_r878653229
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/TopicLookupBase.java:
##########
@@ -220,49 +220,52 @@ public static CompletableFuture<ByteBuf>
lookupTopicAsync(PulsarService pulsarSe
cluster);
}
validationFuture.complete(newLookupResponse(differentClusterData.getBrokerServiceUrl(),
- differentClusterData.getBrokerServiceUrlTls(), true,
LookupType.Redirect, requestId, false));
+ differentClusterData.getBrokerServiceUrlTls(), true,
LookupType.Redirect,
+ requestId, false));
} else {
// (2) authorize client
- try {
- checkAuthorization(pulsarService, topicName, clientAppId,
authenticationData);
- } catch (RestException authException) {
- log.warn("Failed to authorized {} on cluster {}",
clientAppId, topicName.toString());
-
validationFuture.complete(newLookupErrorResponse(ServerError.AuthorizationError,
- authException.getMessage(), requestId));
- return;
- } catch (Exception e) {
- log.warn("Unknown error while authorizing {} on cluster
{}", clientAppId, topicName.toString());
- validationFuture.completeExceptionally(e);
- return;
- }
- // (3) validate global namespace
- checkLocalOrGetPeerReplicationCluster(pulsarService,
topicName.getNamespaceObject())
- .thenAccept(peerClusterData -> {
- if (peerClusterData == null) {
- // (4) all validation passed: initiate lookup
- validationFuture.complete(null);
- return;
- }
- // if peer-cluster-data is present it means
namespace is owned by that peer-cluster and
- // request should be redirect to the peer-cluster
- if
(StringUtils.isBlank(peerClusterData.getBrokerServiceUrl())
- &&
StringUtils.isBlank(peerClusterData.getBrokerServiceUrlTls())) {
-
validationFuture.complete(newLookupErrorResponse(ServerError.MetadataError,
- "Redirected cluster's brokerService
url is not configured", requestId));
- return;
+ checkAuthorizationAsync(pulsarService, topicName, clientAppId,
authenticationData).thenRun(() ->
+ validationFuture.complete(null))
+ .exceptionally(e -> {
+ if (e.getCause() instanceof RestException) {
Review Comment:
```suggestion
if (FutureUtil.unwrapCompletionException(e)
instanceof RestException) {
```
--
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]
