@Override
    public RemotingCommand processRequest(final ChannelHandlerContext ctx,
        RemotingCommand request) throws RemotingCommandException {
        return this.processRequest(ctx.channel(), request, true);
    }
@Test
    public void testProcessRequest_NoPermission() throws RemotingCommandException {
        this.brokerController.getBrokerConfig().setBrokerPermission(PermName.PERM_WRITE);
        RemotingCommand request = createPeekMessageRequest("group","topic",0);
        RemotingCommand response = peekMessageProcessor.processRequest(handlerContext, request);
        assertThat(response.getCode()).isEqualTo(ResponseCode.NO_PERMISSION);
        this.brokerController.getBrokerConfig().setBrokerPermission(PermName.PERM_WRITE | PermName.PERM_READ);

        topicConfigManager.getTopicConfigTable().get("topic").setPerm(PermName.PERM_WRITE);
        response = peekMessageProcessor.processRequest(handlerContext, request);
        assertThat(response.getCode()).isEqualTo(ResponseCode.NO_PERMISSION);
        topicConfigManager.getTopicConfigTable().get("topic").setPerm(PermName.PERM_WRITE | PermName.PERM_READ);

        when(subscriptionGroupConfig.isConsumeEnable()).thenReturn(false);
        response = peekMessageProcessor.processRequest(handlerContext, request);
        assertThat(response.getCode()).isEqualTo(ResponseCode.NO_PERMISSION);
    }