handle unauthorized

apache · Oct 19, 2022 · e34524a · e34524a
1 parent b199d7b
commit e34524a
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 2 deletions.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerServiceException.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerServiceException.java
@@ -204,6 +204,12 @@ public TopicBacklogQuotaExceededException(BacklogQuota.RetentionPolicy retention
         }
     }
 
+    public static class UnauthorizedException extends BrokerServiceException {
+        public UnauthorizedException(String msg) {
+            super(msg);
+        }
+    }
+
     public static org.apache.pulsar.common.api.proto.ServerError getClientErrorCode(Throwable t) {
         return getClientErrorCode(t, true);
     }

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
@@ -1108,7 +1108,8 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
                                     consumers.remove(consumerId, consumerFuture);
                                     ctx.writeAndFlush(Commands.newError(requestId, ServerError.AuthorizationError,
                                             msg));
-                                    return null;
+                                    return FutureUtil
+                                            .failedFuture(new BrokerServiceException.UnauthorizedException(msg));
                                 }
                             }
 
@@ -1340,7 +1341,8 @@ protected void handleProducer(final CommandProducer cmdProducer) {
                         log.warn("[{}] {} with role {}", remoteAddress, msg, getPrincipal());
                         ctx.writeAndFlush(Commands.newError(requestId, ServerError.AuthorizationError,
                                 msg));
-                        return null;
+                        return FutureUtil
+                                .failedFuture(new BrokerServiceException.UnauthorizedException(msg));
                     }
                 }