Return correct authz and auth errors from proxy to client (apache#9055)

Co-authored-by: Jerry Peng <[email protected]>
xche · Dec 26, 2020 · 8574e58 · 8574e58
1 parent 2c3e33b
commit 8574e58
Showing 1 changed file with 15 additions and 2 deletions.
diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/LookupProxyHandler.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/LookupProxyHandler.java
@@ -27,6 +27,7 @@
 import java.util.Optional;
 
 import org.apache.commons.lang3.StringUtils;
+import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.common.protocol.Commands;
 import org.apache.pulsar.common.api.proto.PulsarApi.CommandGetTopicsOfNamespace;
 import org.apache.pulsar.common.api.proto.PulsarApi.CommandGetSchema;
@@ -231,7 +232,7 @@ private void handlePartitionMetadataResponse(CommandPartitionedTopicMetadata par
                         log.warn("[{}] Failed to get partitioned metadata for topic {} {}", clientAddress, topicName,
                                 ex.getMessage(), ex);
                         proxyConnection.ctx().writeAndFlush(Commands.newPartitionMetadataResponse(
-                                ServerError.ServiceNotReady, ex.getMessage(), clientRequestId));
+                          getServerError(ex), ex.getMessage(), clientRequestId));
                         return null;
                     });
         } else {
@@ -259,7 +260,7 @@ private void handlePartitionMetadataResponse(CommandPartitionedTopicMetadata par
                     if (t != null) {
                         log.warn("[{}] failed to get Partitioned metadata : {}", topicName.toString(),
                             t.getMessage(), t);
-                        proxyConnection.ctx().writeAndFlush(Commands.newLookupErrorResponse(ServerError.ServiceNotReady,
+                        proxyConnection.ctx().writeAndFlush(Commands.newLookupErrorResponse(getServerError(t),
                             t.getMessage(), clientRequestId));
                     } else {
                         proxyConnection.ctx().writeAndFlush(
@@ -443,5 +444,17 @@ private InetSocketAddress getAddr(String brokerServiceUrl, long clientRequestId)
         return InetSocketAddress.createUnresolved(brokerURI.getHost(), brokerURI.getPort());
     }
 
+    private ServerError getServerError(Throwable error) {
+        ServerError responseError;
+        if (error instanceof PulsarClientException.AuthorizationException) {
+            responseError = ServerError.AuthorizationError;
+        } else if (error instanceof PulsarClientException.AuthenticationException) {
+            responseError = ServerError.AuthenticationError;
+        } else {
+            responseError = ServerError.ServiceNotReady;
+        }
+        return responseError;
+    }
+
     private static final Logger log = LoggerFactory.getLogger(LookupProxyHandler.class);
 }