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
index 6d0473ed99c66..a297b437b1f7d 100644
--- 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
@@ -634,16 +634,6 @@ ByteBuf createConsumerStatsResponse(Consumer consumer, long requestId) {
 
     // complete the connect and sent newConnected command
     private void completeConnect(int clientProtoVersion, String clientVersion, boolean supportsTopicWatchers) {
-        if (service.isAuthenticationEnabled() && service.isAuthorizationEnabled()) {
-            if (!service.getAuthorizationService()
-                    .isValidOriginalPrincipal(authRole, originalPrincipal, remoteAddress)) {
-                state = State.Failed;
-                service.getPulsarStats().recordConnectionCreateFail();
-                final ByteBuf msg = Commands.newError(-1, ServerError.AuthorizationError, "Invalid roles.");
-                ctx.writeAndFlush(msg).addListener(ChannelFutureListener.CLOSE);
-                return;
-            }
-        }
         ctx.writeAndFlush(Commands.newConnected(clientProtoVersion, maxMessageSize, supportsTopicWatchers));
         state = State.Connected;
         service.getPulsarStats().recordConnectionCreateSuccess();
@@ -660,7 +650,8 @@ private void completeConnect(int clientProtoVersion, String clientVersion, boole
     }
 
     // According to auth result, send newConnected or newAuthChallenge command.
-    private State doAuthentication(AuthData clientData,
+    private void doAuthentication(AuthData clientData,
+                                   boolean useOriginalAuthState,
                                    int clientProtocolVersion,
                                    String clientVersion) throws Exception {
 
@@ -668,8 +659,7 @@ private State doAuthentication(AuthData clientData,
         // in presence of a proxy and if the proxy is forwarding the credentials).
         // In this case, the re-validation needs to be done against the original client
         // credentials.
-        boolean useOriginalAuthState = (originalAuthState != null);
-        AuthenticationState authState =  useOriginalAuthState ? originalAuthState : this.authState;
+        AuthenticationState authState = useOriginalAuthState ? originalAuthState : this.authState;
         String authRole = useOriginalAuthState ? originalPrincipal : this.authRole;
         AuthData brokerData = authState.authenticate(clientData);
 
@@ -702,6 +692,16 @@ private State doAuthentication(AuthData clientData,
 
             if (state != State.Connected) {
                 // First time authentication is done
+                if (service.isAuthenticationEnabled() && service.isAuthorizationEnabled()) {
+                    if (!service.getAuthorizationService()
+                            .isValidOriginalPrincipal(this.authRole, originalPrincipal, remoteAddress)) {
+                        state = State.Failed;
+                        service.getPulsarStats().recordConnectionCreateFail();
+                        final ByteBuf msg = Commands.newError(-1, ServerError.AuthorizationError, "Invalid roles.");
+                        ctx.writeAndFlush(msg).addListener(ChannelFutureListener.CLOSE);
+                        return;
+                    }
+                }
                 completeConnect(clientProtocolVersion, clientVersion, enableSubscriptionPatternEvaluation);
             } else {
                 // If the connection was already ready, it means we're doing a refresh
@@ -715,18 +715,16 @@ private State doAuthentication(AuthData clientData,
                     }
                 }
             }
+        } else {
 
-            return State.Connected;
-        }
-
-        // auth not complete, continue auth with client side.
-        ctx.writeAndFlush(Commands.newAuthChallenge(authMethod, brokerData, clientProtocolVersion));
-        if (log.isDebugEnabled()) {
-            log.debug("[{}] Authentication in progress client by method {}.",
-                remoteAddress, authMethod);
-            log.debug("[{}] connect state change to : [{}]", remoteAddress, State.Connecting.name());
+            // auth not complete, continue auth with client side.
+            ctx.writeAndFlush(Commands.newAuthChallenge(authMethod, brokerData, clientProtocolVersion));
+            if (log.isDebugEnabled()) {
+                log.debug("[{}] Authentication in progress client by method {}.",
+                        remoteAddress, authMethod);
+                log.debug("[{}] connect state change to : [{}]", remoteAddress, State.Connecting.name());
+            }
         }
-        return State.Connecting;
     }
 
     public void refreshAuthenticationCredentials() {
@@ -824,6 +822,8 @@ protected void handleConnect(CommandConnect connect) {
             return;
         }
 
+        state = State.Connecting;
+
         try {
             byte[] authData = connect.hasAuthData() ? connect.getAuthData() : emptyArray;
             AuthData clientData = AuthData.of(authData);
@@ -871,8 +871,6 @@ protected void handleConnect(CommandConnect connect) {
                 log.debug("[{}] Authenticate role : {}", remoteAddress, role);
             }
 
-            state = doAuthentication(clientData, clientProtocolVersion, clientVersion);
-
             // This will fail the check if:
             //  1. client is coming through a proxy
             //  2. we require to validate the original credentials
@@ -914,6 +912,7 @@ protected void handleConnect(CommandConnect connect) {
                         remoteAddress, originalPrincipal);
                 }
             }
+            doAuthentication(clientData, false, clientProtocolVersion, clientVersion);
         } catch (Exception e) {
             service.getPulsarStats().recordConnectionCreateFail();
             logAuthException(remoteAddress, "connect", getPrincipal(), Optional.empty(), e);
@@ -937,7 +936,7 @@ protected void handleAuthResponse(CommandAuthResponse authResponse) {
 
         try {
             AuthData clientData = AuthData.of(authResponse.getResponse().getAuthData());
-            doAuthentication(clientData, authResponse.getProtocolVersion(),
+            doAuthentication(clientData, originalAuthState != null, authResponse.getProtocolVersion(),
                     authResponse.hasClientVersion() ? authResponse.getClientVersion() : EMPTY);
         } catch (AuthenticationException e) {
             service.getPulsarStats().recordConnectionCreateFail();