private CompletableFuture<Boolean> isSuperUser() {
        assert ctx.executor().inEventLoop();
        if (service.isAuthenticationEnabled() && service.isAuthorizationEnabled()) {
            CompletableFuture<Boolean> isAuthRoleAuthorized = service.getAuthorizationService().isSuperUser(
                    authRole, authenticationData);
            if (originalPrincipal != null) {
                CompletableFuture<Boolean> isOriginalPrincipalAuthorized = service.getAuthorizationService()
                        .isSuperUser(originalPrincipal,
                                originalAuthData != null ? originalAuthData : authenticationData);
                return isOriginalPrincipalAuthorized.thenCombine(isAuthRoleAuthorized,
                        (originalPrincipal, authRole) -> originalPrincipal && authRole);
            } else {
                return isAuthRoleAuthorized;
            }
        } else {
            return CompletableFuture.completedFuture(true);
        }
    }
@Test(timeOut = 30000)
    public void testClusterAccess() throws Exception {
        svcConfig.setAuthorizationEnabled(true);
        AuthorizationService authorizationService =
                spyWithClassAndConstructorArgs(AuthorizationService.class, svcConfig, pulsar.getPulsarResources());
        Field providerField = AuthorizationService.class.getDeclaredField("provider");
        providerField.setAccessible(true);
        PulsarAuthorizationProvider authorizationProvider =
                spyWithClassAndConstructorArgs(PulsarAuthorizationProvider.class, svcConfig,
                        pulsar.getPulsarResources());
        providerField.set(authorizationService, authorizationProvider);
        doReturn(authorizationService).when(brokerService).getAuthorizationService();
        svcConfig.setAuthorizationEnabled(true);
        doReturn(CompletableFuture.completedFuture(false)).when(authorizationProvider)
                .isSuperUser(Mockito.anyString(), Mockito.any(), Mockito.any());
        doReturn(CompletableFuture.completedFuture(false)).when(authorizationProvider)
                .validateTenantAdminAccess(Mockito.anyString(), Mockito.any(), Mockito.any());
        doReturn(CompletableFuture.completedFuture(true)).when(authorizationProvider)
                .checkPermission(any(TopicName.class), Mockito.anyString(),
                        any(AuthAction.class));

        resetChannel();
        setChannelConnected();
        ByteBuf clientCommand = Commands.newProducer(successTopicName, 1 /* producer id */, 1 /* request id */,
                "prod-name", Collections.emptyMap(), false);
        channel.writeInbound(clientCommand);
        assertTrue(getResponse() instanceof CommandProducerSuccess);

        resetChannel();
        setChannelConnected();
        clientCommand = Commands.newProducer(topicWithNonLocalCluster, 1 /* producer id */, 1 /* request id */,
                "prod-name", Collections.emptyMap(), false);
        channel.writeInbound(clientCommand);
        assertTrue(getResponse() instanceof CommandError);
        channel.finish();
    }