[fix][broker]Fixed produce and consume when anonymousUserRole enabled (…

…#21237)

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java

@@ -863,7 +863,6 @@ protected void handleConnect(CommandConnect connect) {
         try {
             byte[] authData = connect.hasAuthData() ? connect.getAuthData() : emptyArray;
             AuthData clientData = AuthData.of(authData);
-
             // init authentication
             if (connect.hasAuthMethodName()) {
                 authMethod = connect.getAuthMethodName();
@@ -924,10 +923,22 @@ protected void handleConnect(CommandConnect connect) {
                         .getAuthenticationService()
                         .getAuthenticationProvider(originalAuthMethod);
 
+                /**
+                 * When both the broker and the proxy are configured with anonymousUserRole
+                 * if the client does not configure an authentication method
+                 * the proxy side will set the value of anonymousUserRole to clientAuthRole when it creates a connection
+                 * and the value of clientAuthMethod will be none.
+                 * Similarly, should also set the value of authRole to anonymousUserRole on the broker side.
+                 */
                 if (originalAuthenticationProvider == null) {
-                    throw new AuthenticationException(
-                            String.format("Can't find AuthenticationProvider for original role"
-                                    + " using auth method [%s] is not available", originalAuthMethod));
+                    authRole = getBrokerService().getAuthenticationService().getAnonymousUserRole()
+                            .orElseThrow(() ->
+                                    new AuthenticationException("No anonymous role, and can't find "
+                                            + "AuthenticationProvider for original role using auth method "
+                                            + "[" + originalAuthMethod + "] is not available"));
+                    originalPrincipal = authRole;
+                    completeConnect(clientProtocolVersion, clientVersion);
+                    return;
                 }
 
                 AuthData originalAuthDataCopy =  AuthData.of(connect.getOriginalAuthData().getBytes());

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java

@@ -504,6 +504,41 @@ public void testConnectCommandWithPassingOriginalAuthData() throws Exception {
         channel.finish();
     }
 
+    @Test(timeOut = 30000)
+    public void testConnectCommandWithPassingOriginalAuthDataAndSetAnonymousUserRole() throws Exception {
+        AuthenticationService authenticationService = mock(AuthenticationService.class);
+        AuthenticationProvider authenticationProvider = new MockAuthenticationProvider();
+        String authMethodName = authenticationProvider.getAuthMethodName();
+
+        String anonymousUserRole = "admin";
+        when(brokerService.getAuthenticationService()).thenReturn(authenticationService);
+        when(authenticationService.getAuthenticationProvider(authMethodName)).thenReturn(authenticationProvider);
+        when(authenticationService.getAnonymousUserRole()).thenReturn(Optional.of(anonymousUserRole));
+        svcConfig.setAuthenticationEnabled(true);
+        svcConfig.setAuthenticateOriginalAuthData(true);
+        svcConfig.setProxyRoles(Collections.singleton("pass.proxy"));
+        svcConfig.setAnonymousUserRole(anonymousUserRole);
+
+        resetChannel();
+        assertTrue(channel.isActive());
+        assertEquals(serverCnx.getState(), State.Start);
+
+        // When both the proxy and the broker set the anonymousUserRole option
+        // the proxy will use anonymousUserRole to delegate the client's role when connecting.
+        ByteBuf clientCommand = Commands.newConnect(authMethodName, "pass.proxy", 1, null,
+                null, anonymousUserRole, null, null);
+        channel.writeInbound(clientCommand);
+
+        Object response1 = getResponse();
+        assertTrue(response1 instanceof CommandConnected);
+        assertEquals(serverCnx.getState(), State.Connected);
+        assertEquals(serverCnx.getAuthRole(), anonymousUserRole);
+        assertEquals(serverCnx.getPrincipal(), anonymousUserRole);
+        assertEquals(serverCnx.getOriginalPrincipal(), anonymousUserRole);
+        assertTrue(serverCnx.isActive());
+        channel.finish();
+    }
+
     @Test(timeOut = 30000)
     public void testConnectCommandWithPassingOriginalPrincipal() throws Exception {
         AuthenticationService authenticationService = mock(AuthenticationService.class);