Return correct authz and auth errors from proxy to client
@@ -27,6 +27,7 @@ import java.net.URISyntaxException;
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 @@ public class LookupProxyHandler {
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 @@ public class LookupProxyHandler {
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 @@ public class LookupProxyHandler {
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);
}
